当前位置: 首页 > 面试题库 >

芹菜:如何限制队列中的任务数量并在满时停止进食?

谢旭
2023-03-14
问题内容

我是Celery的新手,这是我的问题:

假设我有一个脚本,该脚本经常应该从数据库中获取新数据并将其发送给使用Celery的工作人员。

task.py

# Celery Task
from celery import Celery

app = Celery('tasks', broker='amqp://guest@localhost//')

@app.task
def process_data(x):
    # Do something with x
    pass

fetch_db.py

# Fetch new data from DB and dispatch to workers.
from tasks import process_data

while True:
    # Run DB query here to fetch new data from DB fetched_data

    process_data.delay(fetched_data)

    sleep(30);

我担心的是:每30秒获取一次数据。process_data()函数可能需要更长的时间,并且根据工作人员的数量(尤其是如果数量太少),据我所知,队列可能会受到限制。

  1. 我不能增加工人的数量。
  2. 我可以修改代码,以免在队列满时提供队列。

问题是如何设置队列大小以及如何知道队列已满?一般来说,如何处理这种情况?


问题答案:

您可以设置的RabbitMQx-max-length
使用队列预先声明海带

例如:

import time
from celery import Celery
from kombu import Queue, Exchange

class Config(object):
    BROKER_URL = "amqp://guest@localhost//"

    CELERY_QUEUES = (
        Queue(
            'important',
            exchange=Exchange('important'),
            routing_key="important",
            queue_arguments={'x-max-length': 10}
        ),
    )

app = Celery('tasks')
app.config_from_object(Config)


@app.task(queue='important')
def process_data(x):
    pass

或使用政策

rabbitmqctl set_policy Ten "^one-meg$" '{"max-length-bytes":1000000}' --apply-to queues


 类似资料:
  • 我已经安装了以下版本的芹菜和拉比MQ- 芹菜3.1。6 rabbitmq 3.1。1. 我可以从PHP向默认队列发布任务- 我的worker模块是用python编写的- 我像这样管理芹菜工人和客户- 这很有效。我在终端窗口1中看到以下输出: 但是我想要不同的队列。为了演示,假设我只需要一个名为demo的队列。所以我像这样管理我的芹菜工人- 但它不起作用。任务未得到执行。我猜这可能是因为PHP代码将

  • 问题内容: 我将向芹菜队列添加多个任务,然后等待结果。我有各种各样的想法,我将如何利用某种形式的共享存储(memcached,redis,db等)来实现这一目标,但是,我本以为Celery可以自动处理,但我无法在线找到任何资源。 代码示例 问题答案: 对于 芹菜 > = 3.0 ,使用taskset被弃用赞成组。 在后台启动组: 等待:

  • 问题内容: 有没有一种方法可以停止执行无限循环的线程? 问题答案: 是的,您可以将替换(或在逻辑上)。 这样,当任务取消时,循环将终止。 循环看起来像这样: 使用应该是这样的:

  • 问题内容: 我研究了,但是示例仅涉及使其重现。我正在寻找一种类似的功能,例如说“每个星期一的凌晨1点执行此任务”的功能。 问题答案: 感谢Patrick Altman,最近发布的1.0.3版现在支持此功能。 例: 请参阅更改日志以获取更多信息: http://celeryproject.org/docs/changelog.html

  • 所以我使用executorservice创建了一个线程池。 我试图访问线程池队列中的任务数。我看到没有办法得到它。我知道有一些方法来获取线程池执行器中的队列大小,但是我如何使用执行器服务对象来实现这一点。 就像我说的,如果我创建了一个像这样的线程池执行器,我可以得到队列信息 我知道我可以使用tpExecutor。队列size()获取线程池队列中的任务数。但目前我已经使用Executor服务声明了我

  • 问题内容: 我正在使用类,并且正在使用其schedule方法执行某些任务,但是在执行了6次之后,我必须停止其任务。 我该怎么办? 问题答案: 在某个地方保留对计时器的引用,并使用: 停止所做的一切。您可以将此代码放在要执行的任务中,static int以计算您经过的次数,例如