在 Celery 中的组中设置任务之间的延迟

Question

我有一个 python 应用程序，用户可以在其中启动某个任务。

任务的全部目的也是执行给定数量的 POST/GET 请求，并以特定的时间间隔到给定的 URL。

所以用户给出 N - 请求数，V - 每秒请求数。

考虑到由于 I/O 延迟，实际 r/s 速度可能更大或更小，设计这样的任务如何更好。

首先，我决定将 Celery 与 Eventlet 一起使用，否则我将需要大量无法接受的作品。

我天真的方法：

• 客户端使用 task.delay() 启动任务

• 内部任务我做这样的事情：

  @task
  def task(number_of_requests, time_period):
     for _ in range(number_of_requests):
         start = time.time()
         params_for_concrete_subtask = ...
         # .... do some IO with monkey_patched eventlet requests library
         elapsed = (time.time() - start)
         # If we completed this subtask to fast
         if elapsed < time_period / number_of_requests:
             eventlet.sleep(time_period / number_of_requests)
  

一个工作示例是here。

如果我们太快，我们会尝试等待以保持所需的速度。如果我们太慢，从客户的角度来看是可以的。我们不违反请求/第二个要求。但是，如果我重新启动 Celery，这会正确恢复吗？

我认为这应该有效，但我认为有更好的方法。在 Celery 中，我可以定义一个具有特定速率限制的任务，这几乎符合我的需求保证。所以我可以使用 Celerygroup功能并编写：

@task(rate_limit=...)
def task(...):
    #

task_executor = task.s(number_of_requests, time_period)
group(task_executor(params_for_concrete_task) for params_for_concrete_task in ...).delay()

但是在这里我对动态的 rate_limit 进行了硬编码，我没有看到改变它的方法。我看到了一个例子：

  task.s(....).set(... params ...)

但是我试图传递rate_limit给set它不起作用的方法。

另一个可能更好的想法是使用 Celery 的周期性任务调度程序。默认的执行周期和要定期执行的任务是固定的。

我需要能够动态创建任务，这些任务以特定的速率限制定期运行给定的次数。也许我需要运行我自己的调度程序来从数据库中获取任务？但我没有看到任何关于此的文档。

另一种方法是尝试使用chain函数，但我无法弄清楚任务参数之间是否存在延迟。

Answer 1

如果要动态调整 rate_limit，可以使用以下代码进行。它还在运行时创建 chain()。运行这个你会看到我们成功地将 5/sec 的 rate_limit 覆盖到 0.5/sec。

测试任务.py

from celery import Celery, signature, chain
import datetime as dt

app = Celery('test_tasks')
app.config_from_object('celery_config')

@app.task(bind=True, rate_limit=5)
def test_1(self):
    print dt.datetime.now()


app.control.broadcast('rate_limit',
                       arguments={'task_name': 'test_tasks.test_1',
                                  'rate_limit': 0.5})

test_task = signature('test_tasks.test_1').set(immutable=True)

l = [test_task] * 100

chain = chain(*l)
res = chain()

我还尝试从类中覆盖该属性，但是 IMO 在工作人员注册任务时设置了 rate_limit，这就是 .set() 没有效果的原因。我在这里推测，必须检查源代码。

解决方案2

使用前一个调用的结束时间实现你自己的等待机制，在链中函数的返回被传递到下一个。

所以它看起来像这样：

from celery import Celery, signature, chain
import datetime as dt
import time

app = Celery('test_tasks')
app.config_from_object('celery_config')

@app.task(bind=True)
def test_1(self, prev_endtime=dt.datetime.now(), wait_seconds=5):
    wait = dt.timedelta(seconds=wait_seconds)
    print dt.datetime.now() - prev_endtime
    wait = wait - (dt.datetime.now() - prev_endtime)
    wait = wait.seconds
    print wait
    time.sleep(max(0, wait))
    now = dt.datetime.now()
    print now
    return now

#app.control.rate_limit('test_tasks.test_1', '0.5')
test_task = signature('test_tasks.test_1')

l = [test_task] * 100

chain = chain(*l)
res = chain()

我认为这实际上比广播更可靠。