5

我正在尝试发送约 400 个 HTTP GET 请求并收集结果。我正在从 django 运行。我的解决方案是将 celery 与 gevent 一起使用。

要启动 celery 任务，我调用get_reports：

def get_reports(self, clients, *args, **kw):
    sub_tasks = []
    for client in clients:  
            s = self.get_report_task.s(self, client, *args, **kw).set(queue='io_bound')
        sub_tasks.append(s)
    res = celery.group(*sub_tasks)()
    reports = res.get(timeout=30, interval=0.001)
    return reports

@celery.task
def get_report_task(self, client, *args, **kw):
    report = send_http_request(...)
    return report

我使用 4 名工人：

manage celery worker -P gevent --concurrency=100 -n a0 -Q io_bound
manage celery worker -P gevent --concurrency=100 -n a1 -Q io_bound
manage celery worker -P gevent --concurrency=100 -n a2 -Q io_bound
manage celery worker -P gevent --concurrency=100 -n a3 -Q io_bound

我使用 RabbitMq 作为代理。

虽然它比按顺序运行请求快得多（400 个请求大约需要 23 秒），但我注意到大部分时间都是 celery 本身的开销，即如果我像这样更改get_report_task：

@celery.task
def get_report_task(self, client, *args, **kw):
    return []

整个操作耗时约 19 秒。这意味着我只花了 19 秒将所有任务发送到 celery 并返回结果

到rabbit mq的消息排队率似乎绑定到28条消息/秒，我认为这是我的瓶颈。

如果这很重要，我正在一台win 8机器上运行。

我尝试过的一些事情：

  • 使用 redis 作为代理
  • 使用 redis 作为结果后端

  • 使用这些设置进行调整

    BROKER_POOL_LIMIT = 500

    CELERYD_PREFETCH_MULTIPLIER = 0

    CELERYD_MAX_TASKS_PER_CHILD = 100

    CELERY_ACKS_LATE = 假

    CELERY_DISABLE_RATE_LIMITS = 真

我正在寻找任何有助于加快速度的建议。

4

3 回答 3

7

您真的在没有虚拟机的情况下在 Windows 8 上运行吗？我在运行 OS X 10.7 的 2 核 Macbook 8GB RAM 上进行了以下简单测试：

import celery
from time import time

@celery.task
def test_task(i):
    return i

grp = celery.group(test_task.s(i) for i in range(400))
tic1 = time(); res = grp(); tac1 = time()
print 'queued in', tac1 - tic1
tic2 = time(); vals = res.get(); tac2 = time()
print 'executed in', tac2 - tic2

我使用 Redis 作为代理，使用 Postgres 作为后端，使用--concurrency=4. 猜猜输出是什么？这里是：

排队 3.5009469986

执行于 2.99818301201

于 2013-09-18T10:26:57.053 回答
3

好吧，事实证明我有两个不同的问题。

首先，任务是一个成员方法。将其从课堂中提取出来后，时间降到了12秒左右。我只能假设它与self的酸洗有关。

第二件事是它在 Windows 上运行。在我的 linux 机器上运行后，运行时间不到 2 秒。猜猜窗户不是为了高性能而切割的。

于 2013-09-24T20:44:01.300 回答
1

用twisted代替怎么样？您可以获得更简单的应用程序结构。您可以一次从 django 进程发送所有 400 个请求，然后等待所有请求完成。这同时起作用，因为 twisted 将套接字设置为非阻塞模式，并且仅在数据可用时读取数据。

前段时间我遇到了类似的问题，我在 twisted 和 django 之间建立了一个很好的桥梁。我在生产环境中运行它已经快一年了。你可以在这里找到它：https ://github.com/kowalski/featdjango/ 。简而言之，它具有运行主扭曲反应器循环的主应用程序线程，并且 django 视图结果被委托给一个线程。它使用一个特殊的线程池，它公开了与反应器交互并使用其异步功能的方法。

如果您使用它，您的代码将如下所示：

from twisted.internet import defer
from twisted.web.client import getPage

import threading


def get_reports(self, urls, *args, **kw):
    ct = threading.current_thread()

    defers = list()
    for url in urls:
        # here the Deferred is created which will fire when
        # the call is complete
        d = ct.call_async(getPage, args=[url] + args, kwargs=kw)
        # here we keep it for reference
        defers.append(d)

    # here we create a Deferred which will fire when all the
    # consiting Deferreds are completed
    deferred_list = defer.DeferredList(defers, consumeErrors=True)
    # here we tell the current thread to wait until we are done
    results = ct.wait_for_defer(deferred_list)

    # the results is a list of the form (C{bool} success flag, result)
    # below unpack it
    reports = list()
    for success, result in results:
        if success:
            reports.append(result)
        else:
            # here handle the failure, or just ignore
            pass

    return reports

这仍然是您可以优化的东西。在这里，每次调用 getPage() 都会创建一个单独的 TCP 连接，并在完成后关闭它。如果您的 400 个请求中的每一个都发送到不同的主机，那么这是尽可能优化的。如果不是这种情况，您可以使用 http 连接池，它使用持久连接和 http 管道。你像这样实例化它：

from feat.web import httpclient

pool = httpclient.ConnectionPool(host, port, maximum_connections=3)

不是像这样执行单个请求（这代替了 getPage() 调用）：

d = ct.call_async(pool.request, args=(method, path, headers, body))
于 2013-09-17T08:59:50.637 回答