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

Question

我想在多个python进程之间共享blockingchannel。以便从其他python进程发送basic_ack。

如何跨多个python进程共享blockingchannel。

代码如下：

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

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

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

Answer 1

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

经验法则是每个进程1个连接和每个线程1个通道。

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

null

如果您希望将消息消耗与多处理结合在一起，通常的模式是让主进程接收消息，将它们的有效负载传递给一个工作进程池，并在完成后对它们进行确认。

使用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消费者预取参数控制并发处理多少消息。查看pika.basic_qos查看如何在Python中执行此操作。