python - rabbitmq 队列填满了 celery 任务

标签 python multithreading rabbitmq celery mnesia

我正在使用 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 = 要调用的站点数

重置rabbitmq后第一次运行应用程序时,它添加了n+1个队列

如何才能使这些任务/队列不会持续存在?一旦获得结果,我就不再需要该任务。

task.forget() 失败,并显示 NotImplementedError('backend does notimplementforget.')

任务过期设置似乎没有效果。我的 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

最佳答案

听起来您不想使用 RabbmitMQ 作为结果后端,而只想作为消息代理。请参阅上一个问题:Queues with random GUID being generated in RabbitMQ server

关于python - rabbitmq 队列填满了 celery 任务,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/28650684/

相关文章:

Android IPC LocalSocket 与 Binder (AIDL)

java - RabbitMQ 示例代码未在 Java 中编译

http - 面向服务的架构 - AMQP 或 HTTP

python - 构建 AIS 消息解码器

python - 如何在 Python 中{pivot|denormalize|manipulate} CSV 表

java - Java异步文件写入PipedOutputStream/PipedInputStream(或Reader/Writer)与BlockingQueue的异同

java - 限制迭代器的线程和/或重用线程?

python - 如果在程序退出前使用 "paste",Tk 仅复制到剪贴板

python - 我如何开始学习 Python for web

reflection - 定义具有非常多消息类型的消息传递域