在多个 python 进程之间共享 RabbitMQ 通道

Question

我想在多个 python 进程中共享 BlockingChannel。为了发送 basic_ack 来自其他 python 进程。

如何在多个 python 进程之间共享 BlockingChannel。

代码如下：

self.__connection__ = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
self.__channel__ = self.__connection__.channel()

我尝试使用 pickle 进行转储，但它不允许转储频道并给出错误 can't pickle select.epoll objects 使用以下代码

filepath = "temp/" + "merger_channel.sav"
pickle.dump(self.__channel__, open(filepath, 'wb'))

目标：

目标是从其他 python 进程的通道发送 basic_ack。

Answer 1

在多个线程之间共享通道是一种反模式，您不太可能设法在进程之间共享它。

经验法则是每个进程 1 connection，每个线程 1 channel。

您可以在以下链接中阅读有关此事的更多信息：

13 common RabbitMQ mistakes
RabbitMQ best practices
This SO线程对RabbitMQ和并发消耗进行了深入分析

如果你想将消息消费与多处理配对，通常的模式是让主进程接收消息，将它们的有效负载传递给工作进程池，并在完成后确认它们。

使用 pika.BlockingChannel 和 concurrent.futures.ProcessPoolExecutor 的简单示例：

def ack_message(channel, delivery_tag, _future):
    """Called once the message has been processed.
    Acknowledge the message to RabbitMQ.
    """
    channel.basic_ack(delivery_tag=delivery_tag)

for message in channel.consume(queue='example'):
    method, properties, body = message

    future = pool.submit(process_message, body)
    # use partial to pass channel and ack_tag to callback function
    ack_message_callback = functools.partial(ack_message, channel, method.delivery_tag)
    future.add_done_callback(ack_message_callback)

上面的循环将无休止地消耗example队列中的消息并将它们提交给进程池。您可以通过 RabbitMQ consumer prefetch parameter. Check pika.basic_qos 控制要同时处理的消息数，以查看如何在 Python.

中执行此操作

在多个 python 进程之间共享 RabbitMQ 通道

Share RabbitMQ channel between multiple python processes

python

rabbitmq

multiprocessing

pika