单PC下Python多进程能否1分钟向RabbitMQ发送100万条消息？

没问题，单台PC完全能做到1分钟内向RabbitMQ发送100万条消息，你的现有代码存在多处性能瓶颈，以下是具体分析和优化方案：

现有代码的性能瓶颈 #

• 重复创建连接与队列声明：每个进程都新建BlockingConnection并重复执行queue_declare，连接建立需要和服务器握手、认证，队列声明也要和服务器交互，这些都是高开销操作，大量重复会严重拖慢速度。
• 同步阻塞客户端限制：pika.BlockingConnection是同步阻塞模型，单进程/线程发送消息时，每发一条都要等网络响应，无法高效利用网络IO资源。
• 循环内冗余计算：每次循环都重复生成时间戳、拼接字符串，这些CPU密集型操作会占用大量时间，积累起来拖慢整体速度。
• 多进程的额外开销：多进程的进程间切换、资源复制成本远高于多线程，对于RabbitMQ这种IO密集型任务，多线程或异步模型更高效。

具体优化方案 #

1. 复用连接与队列声明 #

将队列声明移到全局，所有线程/进程复用同一个连接（或使用连接池），避免重复创建连接和声明队列：

import threading
import pika
import time
from datetime import datetime
import sys

# 全局初始化连接和队列，只执行一次
connection = pika.BlockingConnection(pika.ConnectionParameters(port=5672, virtual_host="test"))
channel = connection.channel()
channel.queue_declare(queue='letterbox')

def publish(no_message, producer):
    print(f"start: {time.ctime(time.time())}")
    # 预生成固定前缀，减少循环内字符串拼接
    producer_prefix = f"{producer}_message_no_"
    body_prefix = "aa cccc "
    
    for i in range(1, int(no_message)):
        sendtime = datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-2]
        body = body_prefix + sendtime
        message_id = producer_prefix + str(i)
        
        _properties = pika.BasicProperties(
            content_type='application/json',
            content_encoding='utf-8',
            message_id=message_id,
            timestamp=int(time.time())
        )
        channel.basic_publish(
            exchange='',
            routing_key='letterbox',
            properties=_properties,
            body=body
        )
    print(f"end: {time.ctime(time.time())}")

def thread_publish(no_publisher, no_message):
    jobs = []
    for i in range(int(no_publisher)):
        # 改用多线程，降低进程切换开销
        thread = threading.Thread(target=publish, args=(no_message, f"test_publisher_no_{i}"))
        jobs.append(thread)
    
    for j in jobs:
        j.start()
    
    for j in jobs:
        j.join()
    
    print("List processing complete.")
    connection.close()

if __name__ == "__main__":
    print('starting ..')
    thread_publish(sys.argv[1], sys.argv[2])

2. 启用批量发送 #

将多条消息攒成一批发送，减少网络交互次数，这是提升吞吐量的关键：

def publish(no_message, producer):
    print(f"start: {time.ctime(time.time())}")
    producer_prefix = f"{producer}_message_no_"
    body_prefix = "aa cccc "
    batch_size = 1000  # 每1000条消息批量发送，可根据实际调整
    messages = []
    
    for i in range(1, int(no_message)):
        sendtime = datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-2]
        body = body_prefix + sendtime
        message_id = producer_prefix + str(i)
        
        _properties = pika.BasicProperties(
            content_type='application/json',
            content_encoding='utf-8',
            message_id=message_id,
            timestamp=int(time.time())
        )
        messages.append(('', 'letterbox', _properties, body))
        
        # 达到批量大小就统一发送
        if len(messages) >= batch_size:
            for msg in messages:
                channel.basic_publish(*msg)
            messages = []
    
    # 发送剩余的不足一批的消息
    if messages:
        for msg in messages:
            channel.basic_publish(*msg)
    
    print(f"end: {time.ctime(time.time())}")

3. 使用异步客户端（推荐） #

改用异步IO客户端aio-pika，能最大化利用网络IO资源，吞吐量远高于同步客户端：

import asyncio
import aio_pika
from datetime import datetime
import sys

async def publish(no_message, producer):
    # 建立异步连接
    connection = await aio_pika.connect_robust(
        "amqp://guest:guest@localhost:5672/test"
    )
    
    async with connection:
        channel = await connection.channel()
        await channel.declare_queue("letterbox")
        
        producer_prefix = f"{producer}_message_no_"
        body_prefix = "aa cccc "
        batch_size = 1000
        messages = []
        
        print(f"start: {datetime.now()}")
        
        for i in range(1, int(no_message)):
            sendtime = datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-2]
            body = body_prefix + sendtime
            message_id = producer_prefix + str(i)
            
            properties = aio_pika.MessageProperties(
                content_type='application/json',
                content_encoding='utf-8',
                message_id=message_id,
                timestamp=int(datetime.now().timestamp())
            )
            message = aio_pika.Message(body=body.encode(), properties=properties)
            messages.append(message)
            
            if len(messages) >= batch_size:
                # 异步批量发送
                await asyncio.gather(*[channel.default_exchange.publish(msg, routing_key="letterbox") for msg in messages])
                messages = []
        
        # 发送剩余消息
        if messages:
            await asyncio.gather(*[channel.default_exchange.publish(msg, routing_key="letterbox") for msg in messages])
        
        print(f"end: {datetime.now()}")

async def thread_publish(no_publisher, no_message):
    tasks = []
    for i in range(int(no_publisher)):
        tasks.append(publish(no_message, f"test_publisher_no_{i}"))
    
    await asyncio.gather(*tasks)
    print("List processing complete.")

if __name__ == "__main__":
    print('starting ..')
    asyncio.run(thread_publish(sys.argv[1], sys.argv[2]))

4. RabbitMQ服务器配置优化 #

• 增大socket缓冲区：在rabbitmq.conf中设置tcp_listen_options.backlog = 4096、tcp_listen_options.sndbuf = 1048576、tcp_listen_options.rcvbuf = 1048576，提升网络传输效率
• 调整内存限制：如果服务器内存充足，将vm_memory_high_watermark.relative设置为0.8，避免RabbitMQ因内存不足触发流控
• 关闭持久化：如果不需要消息持久化，声明队列时设置durable=False，避免磁盘IO开销
• 批量确认发布：如果需要保证消息不丢失，启用publisher_confirms并批量确认，减少确认操作的网络开销

总结 #

通过上述代码优化（复用连接、批量发送、异步客户端）和RabbitMQ服务器配置调整，单台PC完全可以达到1分钟发送100万条消息的目标。

内容的提问来源于stack exchange，提问作者trung