cpu长时间100%加载时处理RabbitMQ心跳答案

【问题标题】：Handling RabbitMQ heartbeats when cpu is loaded 100% for a long timecpu长时间100%加载时处理RabbitMQ心跳
【发布时间】：2022-07-28 11:45:06
【问题描述】：

我在我的 python 应用程序中使用 pika 1.1 和 graph-tool 3.4。它使用来自 RabbitMQ 的任务，然后使用图形工具构建图形，然后运行一些计算。

一些计算，例如介数，会占用大量的cpu功率，使cpu使用率长时间达到100%。有时rabbitmq连接断开，导致任务从头开始。

即使计算在单独的进程中运行，我的猜测是在 cpu 加载 100% 期间，它找不到任何机会向 rabbitmq 发送心跳，这导致连接终止。这并不总是发生，这表明它可能会不时发送心跳。这只是我的猜测，我不确定还有什么可能导致这种情况。

我尝试使用nice(19) 降低计算过程的优先级，但没有成功。我假设它不会影响图形工具生成的进程，它自己并行工作。

由于只是一行代码，graph.calculate_betweenness(... 我没有地方手动发送心跳或减慢执行以创造心跳的机会。

我对因为 cpu 超级忙而未发送心跳的猜测是否正确？
如果是，我该如何处理这种情况？

【问题讨论】：

标签： python-3.x rabbitmq pika graph-tool

【解决方案1】：

回答您的问题：

是的，基本上就是这样。
我们的解决方案是为 CPU 密集型任务创建一个单独的进程。

import time
from multiprocessing import Process

import pika

connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
channel = connection.channel()


channel.exchange_declare(exchange='logs', exchange_type='fanout')
result = channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue
channel.queue_bind(exchange='logs', queue=queue_name)


def cpu_intensive_task(ch, method, properties, body):

    def work(body):
        time.sleep(60)  # If I remember well default HB is 30 seconds
        print(" [x] %r" % body)

    p = Process(target=work, args=(body,))
    p.start()
    # Important to notice if you do p.join() You will have the same problem.

channel.basic_consume(
    queue=queue_name, on_message_callback=cpu_intensive_task, auto_ack=True)

channel.start_consuming()

我想知道这是否是解决这个问题的最佳方案，或者 rabbitMQ 是否是处理 CPU 密集型任务的最佳工具。（对于很长的 CPU 密集型任务（超过 30 分钟），如果您发送手动 ACK，您还需要处理：https://www.rabbitmq.com/consumers.html#acknowledgement-timeout）

【讨论】：