第3章消息中间件部署

3、消息中间件部署：RabbitMQ/Kafka选型对比、单机与集群部署、Topic与Queue设计、消息持久化与可靠性配置

消息中间件，说白了就是多智能体系统的「血管」。

每个智能体都在不停地生产数据、消费数据。没有消息中间件，它们就是一座座孤岛。我见过不少团队，一开始图省事用Redis做消息队列，结果系统一跑起来就崩——嗯，Redis丢消息是出了名的。

3.1 RabbitMQ vs Kafka：选型对比

先聊选型。我个人习惯把这两个中间件看成两种不同的工具：RabbitMQ像快递员，Kafka像仓库管理员。

对比维度	RabbitMQ	Kafka
消息模型	Queue + Exchange	Topic + Partition
吞吐量	万级/秒	百万级/秒
消息顺序	单队列有序	分区内有序
消息回溯	不支持	支持（按offset）
运维复杂度	低	中高
典型场景	任务分发、RPC	日志采集、流处理

为什么多智能体系统我更推荐Kafka？

因为智能体之间的通信往往是「一对多」的广播模式。一个策略智能体发出信号，多个执行智能体都要收到。Kafka的消费者组天然支持这种场景。我曾经在一个项目中用RabbitMQ做广播，结果Exchange绑定关系写了一大堆，运维起来头都大了。

我的建议：

智能体数量少于10个，消息量不大 → RabbitMQ（省心）
智能体数量超过20个，需要历史数据回溯 → Kafka（必须）

3.2 单机部署 vs 集群部署

开发环境用单机，生产环境必须集群。这个道理大家都懂，但怎么搭集群，坑不少。

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.12-management

RabbitMQ集群部署（3节点）

# 节点1
docker run -d --name rabbitmq-1 \
  -p 5672:5672 -p 15672:15672 \
  -e RABBITMQ_ERLANG_COOKIE='secret_cookie' \
  -e RABBITMQ_NODENAME=rabbit@node1 \
  rabbitmq:3.12-management

# 节点2、3加入集群
docker exec rabbitmq-2 rabbitmqctl stop_app
docker exec rabbitmq-2 rabbitmqctl join_cluster rabbit@node1
docker exec rabbitmq-2 rabbitmqctl start_app

注意：RabbitMQ集群不自动同步队列数据。如果节点挂了，该节点上的队列就丢了。我曾经吃过这个亏，后来加了镜像队列模式才解决。

Kafka集群部署（3节点）

# server.properties 关键配置
broker.id=1
listeners=PLAINTEXT://:9092
log.dirs=/data/kafka/logs
zookeeper.connect=zk1:2181,zk2:2181,zk3:2181
num.partitions=3
default.replication.factor=2

Kafka集群的核心是ZooKeeper。没有ZK，Kafka就是一堆散沙。我个人习惯把ZK和Kafka分开部署，ZK用3台，Kafka用3台，互不干扰。

3.3 Topic与Queue设计

设计Topic和Queue，其实就是在规划智能体之间的「对话频道」。

我一般这样设计：

一个智能体类型一个Topic：比如策略智能体发信号到 topic-signal，风控智能体发到 topic-risk
一个消费者组对应一个智能体集群：比如有3个执行智能体，它们属于同一个消费者组，消息只会被其中一个消费
关键信号用单独Queue：比如止损信号，必须单独一个Queue，保证高优先级处理

避坑指南：我曾经把所有信号都塞进一个Topic，结果止损信号被普通信号堵住了，等处理完行情已经过去了3秒。后来我专门给止损信号开了个高优先级Queue，问题才解决。

Kafka Topic设计示例

# 创建Topic
kafka-topics.sh --create \
  --topic trade-signal \
  --partitions 6 \
  --replication-factor 2 \
  --bootstrap-server localhost:9092

# 查看Topic详情
kafka-topics.sh --describe \
  --topic trade-signal \
  --bootstrap-server localhost:9092

分区数怎么定？我有个经验公式：分区数 = 消费者线程数 × 2。比如你有3个执行智能体，每个开2个线程，那就设6个分区。多了浪费，少了会争抢。

3.4 消息持久化与可靠性配置

交易系统最怕丢消息。一条止损消息丢了，可能亏掉几十万。所以持久化和可靠性是重中之重。

RabbitMQ持久化配置

// 生产者：消息标记为持久化
channel.basicPublish(
  exchange,
  routingKey,
  MessageProperties.PERSISTENT_TEXT_PLAIN,  // 关键参数
  message.getBytes()
);

// 队列声明为持久化
channel.queueDeclare("queue-trade", true, false, false, null);

这里有个细节：光设置消息持久化不够，队列本身也要持久化。我见过有人只设了消息持久化，队列没设，结果重启后队列没了，消息全白费。

Kafka持久化配置

# 生产者：acks=all 确保所有副本都写入
props.put("acks", "all");
props.put("retries", 3);
props.put("enable.idempotence", true);

# 消费者：手动提交offset
props.put("enable.auto.commit", "false");

核心配置解读：

acks=all：等所有副本都写入了再返回成功。性能会降一点，但安全
enable.idempotence=true：防止重复消息。交易系统里重复执行订单可是大忌
enable.auto.commit=false：手动提交offset。处理完业务逻辑再提交，确保不丢

消息可靠性架构图

这张图展示了消息从生产者到消费者的完整链路。每个环节都有可靠性保障：生产者发消息时等待ACK，中间件做持久化和副本同步，消费者处理完才提交offset。环环相扣，缺一不可。

我的经验：刚开始做交易系统时，我只配了持久化，没配ACK。结果有一次网络抖动，消息发了但没到，系统也没重试，直接丢了3笔订单。从那以后，我每条消息都配了ACK+重试+死信队列，三层保险。

3.5 总结

消息中间件选型，说白了就是看你的场景。小团队用RabbitMQ，大系统上Kafka。部署一定要集群，持久化一定要配全。

嗯，这些配置看起来琐碎，但每一条都是血泪教训换来的。你想想看，交易系统里一条消息可能就是几万块钱，谁敢马虎？

公众号:蓝海资料掘金营，微信deep3321

📚 课程目录

3、消息中间件部署：RabbitMQ/Kafka选型对比、单机与集群部署、Topic与Queue设计、消息持久化与可靠性配置

3.1 RabbitMQ vs Kafka：选型对比

3.2 单机部署 vs 集群部署

RabbitMQ单机部署（快速上手）

RabbitMQ集群部署（3节点）

Kafka集群部署（3节点）

3.3 Topic与Queue设计

Kafka Topic设计示例

3.4 消息持久化与可靠性配置

RabbitMQ持久化配置

Kafka持久化配置

消息可靠性架构图

3.5 总结