【发布时间】:2015-02-22 06:18:08
【问题描述】:
我正在使用 Celery 通过它们的 IP 地址调用多个硬件单元。每个单元将返回一个值列表。应用代码如下
# create a list of tasks
modbus_calls = []
for site in sites:
call = call_plc.apply_async((site.name, site.address), expires=120) # expires after 2 minutes?
modbus_calls.append(call)
# below checks all tasks are complete (values returned), then move forward out of the while loop
ready_list = [False]
while not all(ready_list):
ready_list = []
for task in modbus_calls:
ready_list.append(task.ready())
# once here, all tasks have returned their values. use the task.get() method to obtain the list of values
在tasks.py文件中,call_plc任务定义为
@app.task
def call_plc(sitename, ip_address):
vals = pc.PLC_Comm().connect_to(sitename, ip_address)
return vals
发生了什么: 在 rabbitmq 开始崩溃(内存不足)之前,我只能运行该应用程序一定次数。我查看/var/lib/rabbitmq/mnesia/rabbit@mymachine/queues,看到一堆带有uuid 名称的队列。这些 uuid 名称与任务 ID 的名称不匹配(从我的应用程序中的 print task.id 学习)。每次运行应用程序时,都会有n 队列添加到此文件夹中,其中n = number of sites to call。
我在重置rabbitmq后第一次运行应用程序时,它添加了n+1队列
我怎样才能使这些任务/队列不会持续存在?一旦我得到结果,我就不再需要该任务了。
task.forget() 失败并显示NotImplementedError('backend does not implement forget.')
任务过期设置似乎没有效果。我的 celeryconfig 文件如下:
BROKER_URL = 'amqp://webdev_rabbit:password@localhost:5672/celeryhost'
CELERY_RESULT_BACKEND = 'amqp://webdev_rabbit:password@localhost:5672/celeryhost'
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_ACCEPT_CONTENT=['json']
CELERY_TIMEZONE = 'Europe/Oslo'
CELERY_ENABLE_UTC = True
CELERY_AMQP_TASK_RESULT_EXPIRES = 120
【问题讨论】:
标签: python multithreading rabbitmq celery mnesia