我是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()函数可能需要更长的时间,并且根据工作人员的数量(尤其是如果数量太少),据我所知,队列可能会受到限制。
问题是如何设置队列大小以及如何知道队列已满?一般来说,如何处理这种情况?
您可以设置的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以计算您经过的次数,例如