RabbitMQ消息队列异步分流实战：优化AI中转接口请求处理

什么时候需要异步分流？

不少AI中转接口系统在直接调用第三方大模型API时，会遇到请求量突增导致响应超时或服务雪崩。
如果每个请求都同步等待模型返回，Tomcat或Gunicorn的工作线程很快被占满，新请求直接被拒绝。
引入RabbitMQ消息队列做异步分流，能有效削峰填谷：生产者把请求塞进队列后立刻返回，消费者从队列拉取任务再调用AI接口，整个过程解耦且可控。

准备工作：搭建RabbitMQ环境

你不需要自己编译，推荐用Docker快速启动一个带管理插件的实例：

docker run -d --name rabbitmq \
  -p 5672:5672 -p 15672:15672 \
  -e RABBITMQ_DEFAULT_USER=admin \
  -e RABBITMQ_DEFAULT_PASS=admin123 \
  rabbitmq:3-management

启动后访问 http://服务器IP:15672，用 admin / admin123 登录。
如果是生产环境，记得修改密码并限制管理端口访问。

下一步创建虚拟主机 /ai_proxy，再建两个交换机：direct_exchange（直接分发）和 topic_exchange（按模型类型路由）。
队列方面，先建 queue.gpt 和 queue.claude，分别绑定到路由键 model.gpt 和 model.claude。
这些操作在管理界面“Queues”和“Exchanges”标签页里完成，或者用代码创建。

核心步骤：生产者发送与消费者接收

生产者：把请求转为消息

假设你的AI中转接口收到请求后，根据 model 字段决定路由。
下面是用Python pika库发送消息的示例：

import pika, json

def send_request(model: str, payload: dict):
    connection = pika.BlockingConnection(pika.ConnectionParameters(
        host='localhost', credentials=pika.PlainCredentials('admin', 'admin123')))
    channel = connection.channel()
    
    # 声明交换机和路由键
    channel.exchange_declare(exchange='ai_exchange', exchange_type='direct', durable=True)
    routing_key = f'model.{model.lower()}'
    
    channel.basic_publish(
        exchange='ai_exchange',
        routing_key=routing_key,
        body=json.dumps(payload),
        properties=pika.BasicProperties(delivery_mode=2)  # 持久化消息
    )
    connection.close()

调用 send_request('gpt', {'prompt': '你好'}) 后，消息会进入 queue.gpt。

消费者：异步拉取并调用AI接口

消费者长期运行，监听队列并处理：

import pika, json, time

def callback(ch, method, properties, body):
    data = json.loads(body)
    print(f'处理模型 {method.routing_key} 的请求: {data}')
    # 这里替换成真实调用AI接口的代码
    time.sleep(1)  # 模拟耗时
    ch.basic_ack(delivery_tag=method.delivery_tag)

connection = pika.BlockingConnection(...)
channel = connection.channel()
channel.basic_qos(prefetch_count=1)  # 一次只取一个任务
channel.basic_consume(queue='queue.gpt', on_message_callback=callback)
channel.start_consuming()

启动多个消费者实例，它们会自动分担队列里的消息，实现真正的高并发。

避坑指南：消息丢失、重复消费与死信队列

• 消息丢失：生产时设置 delivery_mode=2 启用持久化，队列也要声明 durable=True。如果RabbitMQ宕机重启，消息不会丢失。
• 重复消费：消费者处理完一条消息后必须调用 basic_ack 确认。如果没有应答，RabbitMQ会把消息重新投递给其他消费者，导致重复。建议在消费端做幂等处理（如记录请求ID）。
• 死信队列：当消息被拒绝（basic_nack）或超时，可以配置死信交换机（DLX）将失败消息转入 dead_letter_queue，方便后续排查。在创建队列时加上参数：x-dead-letter-exchange=dlx_exchange。

另外，消费者如果抛出异常没有捕获，可能导致消息一直 unack。
最好用 try/finally 确保 ack。

验证一下效果

1. 登录RabbitMQ管理界面，点击“Queues”，查看 queue.gpt 的 Ready 和 Unacked 数量。正常情况 Ready 不应持续上涨。
2. 用生产者发送10条请求，观察消费者日志是否逐条处理，队列积压是否迅速消减。
3. 手动停止一个消费者，再发送请求，消息应该被另一个消费者抢走，不会丢失。

如果发现消息一直卡在 Ready 而消费者没反应，检查 basic_consume 是否绑定了正确的队列，以及 prefetch_count 是否合理（一般设为1）。

当你对RabbitMQ消息队列异步分流流程熟悉后，还可以结合死信队列处理异常、使用TTL控制超时，甚至加上动态路由支持更多模型。
先从本文的步骤跑通基础链路，稳稳落地。