Python - 如何通过 AMQP lib 有效地将 50k 消息推送到 RabbitMQ？

Question

这是我的 MQ 类:

import amqp as amqp

class MQ():

    mqConn = None
    channel = None

    def __init__(self):
        self.connect()

    def connect(self):
        if self.mqConn is None:
            self.mqConn = amqp.Connection(host="localhost", userid="dev", password="dev", virtual_host="/", insist=False)
            self.channel = self.mqConn.channel()

        elif not self.mqConn.connected:
            self.mqConn = amqp.Connection(host="localhost", userid="dev", password="dev", virtual_host="/", insist=False)
            self.channel = self.mqConn.channel()

    def sendMQ(self, message):
        self.connect()
        lMessage = amqp.Message(message)
        self.channel.basic_publish(lMessage, exchange="DevMatrixE", routing_key="dev_matrix_q")

然后我调用 sendMQ() 函数 50,000 次以将消息推送到 RabbitMQ 服务器。完成需要 5 秒多一点。那是每秒大约 10k 条消息，不幸的是这对我的需要来说太慢了。有没有更有效的方法将所有 50,000 条消息都输入服务器？我知道服务器本身可以处理更多的传入消息，但我认为我的瓶颈在 basic_publish 上。

非常感谢。

Answer 1

如果您想在短时间内发送大量消息，使用基于 C 的库 librabbitmq应该能送你到那里。请记住，性能将取决于网络连接的延迟。

我运行了以下非常简单的示例，根据 librabbitmq 发送了 50k 条消息在我的本地 VM 上，它在不到 0.4 秒内完成。

import time

from librabbitmq import Connection

conn = Connection(host="localhost", userid="guest",
                  password="guest", virtual_host="/")
channel = conn.channel()

start_time = time.time()
for _ in xrange(50000):
    channel.basic_publish('This is a test payload!', '', 'simple_queue')

print time.time() - start_time

输出

eandersson@eu-dev-rabbitmq01:~$ python test.py
0.384150981903

我用我自己的库试过了，AMQP-Storm ，但我得到的时间与你使用 py-amqp 的时间差不多，发送 50k 条消息需要 4-5 秒。

另一种解决方案是使用不同的 Python 实现；比如 PyPy。 PyPy 与 Python native AMQP 库一起执行得更好。