第4章 中间件部署：消息队列与数据分发

各位好，欢迎来到中间件部署这一章。说实话，在轨道交通信号系统里，中间件就像人的神经系统——消息传得快不快、准不准，直接决定了整个ATS能不能正常运转。我这些年调试过的项目里，至少有一半的故障都跟中间件配置不当有关。今天咱们就把Kafka、RabbitMQ、DDS和服务注册这些硬骨头啃下来。

4.1 消息队列选型：Kafka vs RabbitMQ

先说说选型。很多新手会问：到底用Kafka还是RabbitMQ？我的回答是：看场景。

Kafka适合高吞吐、日志型、流式处理的场景。比如列车位置数据的采集、信号设备状态日志的汇聚。我在北京某线路项目中，每天要处理上亿条列车位置报文，Kafka扛得住。

RabbitMQ适合低延迟、可靠投递、复杂路由的场景。比如控制命令的下发、报警信息的实时推送。嗯，这里要注意：RabbitMQ的ACK机制很成熟，但吞吐量上限比Kafka低一个数量级。

特性	Kafka	RabbitMQ
吞吐量	百万级/秒	万级/秒
消息持久化	磁盘顺序写，高可靠	内存+磁盘，可靠
路由模式	Topic分区	Exchange绑定
典型场景	日志、轨迹数据	命令、告警

4.2 Kafka安装与调优实战

Kafka的安装其实不复杂，但调优才是关键。我直接给出一套生产级配置。

4.2.1 基础安装

# 下载解压（以3.5.0为例）
wget https://downloads.apache.org/kafka/3.5.0/kafka_2.13-3.5.0.tgz
tar -xzf kafka_2.13-3.5.0.tgz
cd kafka_2.13-3.5.0

# 启动ZooKeeper（Kafka依赖它做协调）
bin/zookeeper-server-start.sh config/zookeeper.properties &

# 启动Kafka
bin/kafka-server-start.sh config/server.properties &

你看，启动就这么几行。但真正上线时，server.properties里的参数才是重头戏。

4.2.2 核心参数调优

我个人习惯把以下参数写在配置文件的头部，方便维护：

# 基础配置
broker.id=1
listeners=PLAINTEXT://0.0.0.0:9092
advertised.listeners=PLAINTEXT://192.168.1.100:9092

# 日志保留策略（轨道交通数据量大，别存太久）
log.retention.hours=72
log.retention.bytes=10737418240  # 10GB

# 性能调优（我踩过坑的地方）
num.network.threads=8
num.io.threads=16
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600

# 分区数建议设置为CPU核心数的2倍
num.partitions=16

4.2.3 生产级Topic创建

# 创建列车位置Topic
bin/kafka-topics.sh --create \
  --bootstrap-server localhost:9092 \
  --replication-factor 3 \
  --partitions 16 \
  --config retention.ms=259200000 \
  --topic train_position

# 创建信号设备状态Topic
bin/kafka-topics.sh --create \
  --bootstrap-server localhost:9092 \
  --replication-factor 3 \
  --partitions 8 \
  --config retention.ms=86400000 \
  --topic signal_status

复制因子设成3，这是生产环境的最低要求。你想想看，如果一台机器挂了，消息丢了，调度员那边看到的数据就是错的——这可不是闹着玩的。

4.3 RabbitMQ安装与配置

RabbitMQ在ATS里主要做命令下发。我习惯把它部署在控制中心，跟Kafka分开。

4.3.1 安装步骤

# Ubuntu/Debian
apt-get install rabbitmq-server

# 启动服务
systemctl start rabbitmq-server
systemctl enable rabbitmq-server

# 启用管理插件（方便运维）
rabbitmq-plugins enable rabbitmq_management

安装完成后，访问http://ip:15672就能看到管理界面。默认账号guest/guest，但生产环境一定要改！

4.3.2 关键配置

嗯，这里要注意：RabbitMQ的配置分两块——Erlang虚拟机参数和RabbitMQ自身参数。

# /etc/rabbitmq/rabbitmq.conf
# 内存阈值（别让RabbitMQ吃掉所有内存）
vm_memory_high_watermark.relative = 0.6

# 磁盘可用空间阈值
disk_free_limit.absolute = 2GB

# 心跳超时（轨道交通网络偶尔抖动，设大点）
heartbeat = 60

# 最大连接数
channel_max = 2048

4.3.3 队列与交换机设计

ATS里我一般用Direct交换机做命令路由，用Topic交换机做告警分发。

# 创建交换机
rabbitmqadmin declare exchange name=cmd.direct type=direct
rabbitmqadmin declare exchange name=alert.topic type=topic

# 创建队列
rabbitmqadmin declare queue name=cmd.station_01 durable=true
rabbitmqadmin declare queue name=cmd.station_02 durable=true

# 绑定
rabbitmqadmin declare binding source=cmd.direct \
  destination=cmd.station_01 routing_key=station_01

说白了，就是让每个车站的ATS客户端只监听自己的队列，互不干扰。这样即使某个车站的网络断了，也不影响其他车站。

4.4 数据分发服务（DDS）配置

DDS在轨道交通里用得越来越多，尤其是实时性要求高的场景——比如列车控制命令的传输。它比Kafka/RabbitMQ更底层，走的是发布-订阅模式，没有中心节点。

4.4.1 DDS核心概念

• DomainParticipant：参与者，每个进程一个
• Topic：数据类型定义，比如列车位置、信号状态
• Publisher/Subscriber：发布者和订阅者
• QoS：服务质量策略，这是DDS的精髓

4.4.2 QoS配置实战

我以RTI Connext DDS为例，给出一个典型的ATS配置：

// 列车位置Topic的QoS配置
struct TrainPositionQos {
  // 可靠性：BEST_EFFORT（丢几个包没关系，位置更新快就行）
  reliability = BEST_EFFORT;
  
  // 历史记录：只保留最新的一条
  history = KEEP_LAST;
  history_depth = 1;
  
  // 时效性：超过100ms的数据直接丢弃
  lifespan = { sec: 0, nanosec: 100000000 };
  
  // 持久性：不持久化（位置数据实时性高）
  durability = VOLATILE;
};

// 控制命令Topic的QoS配置
struct ControlCommandQos {
  // 可靠性：RELIABLE（命令不能丢）
  reliability = RELIABLE;
  
  // 历史记录：保留所有未确认的命令
  history = KEEP_ALL;
  
  // 时效性：30秒内必须送达
  lifespan = { sec: 30, nanosec: 0 };
  
  // 持久性：持久化（防止订阅者晚到）
  durability = TRANSIENT_LOCAL;
};

4.4.3 网络分区配置

轨道交通的网络环境复杂，DDS需要配置发现协议和传输方式：

// 使用UDPv4作为传输层
dds.transport.UDPv4.builtin.parent.message_size_max = 65536

// 发现协议：使用SPDP+SEDP
dds.discovery.enabled = true
dds.discovery.initial_peers = "192.168.1.100,192.168.1.101"

// 忽略某些网段（防止跨区域干扰）
dds.discovery.ignore_loopback_interface = true
dds.discovery.ignore_network_interfaces = "docker0,veth*"

我曾经在某个项目中，DDS的发现报文跑到了Docker网桥上，导致控制中心和车站的参与者互相发现了对方，数据乱套了。加上ignore_network_interfaces就解决了。

4.5 服务注册与发现

ATS系统里服务多得很——数据采集服务、逻辑处理服务、人机界面服务……手动配置IP地址？不现实。服务注册与发现就是干这个的。

4.5.1 选型：Consul vs ZooKeeper vs etcd

组件	优势	劣势	我的推荐
Consul	自带健康检查、KV存储、DNS接口	部署稍重	ATS首选
ZooKeeper	成熟稳定、Kafka依赖它	API较底层	配合Kafka使用
etcd	轻量、Kubernetes原生	功能单一	云原生场景

我个人习惯用Consul，因为它自带的健康检查太方便了。你想想看，一个服务挂了，Consul能自动摘掉它，客户端就不会再往这个节点发请求了。

4.5.2 Consul部署

# 下载Consul
wget https://releases.hashicorp.com/consul/1.16.0/consul_1.16.0_linux_amd64.zip
unzip consul_1.16.0_linux_amd64.zip
mv consul /usr/local/bin/

# 启动Server节点（控制中心）
consul agent -server -bootstrap-expect=3 \
  -data-dir=/var/consul \
  -node=ats-consul-1 \
  -bind=192.168.1.100 \
  -client=0.0.0.0 \
  -ui

# 启动Client节点（车站）
consul agent -data-dir=/var/consul \
  -node=station-consul-01 \
  -bind=192.168.2.10 \
  -join=192.168.1.100

4.5.3 服务注册示例

以Python为例，注册一个列车数据处理服务：

import consul

c = consul.Consul(host='192.168.1.100', port=8500)

# 注册服务
c.agent.service.register(
    name='train-data-processor',
    service_id='train-data-processor-01',
    address='192.168.1.50',
    port=8080,
    tags=['v1', 'production'],
    check=consul.Check().tcp('192.168.1.50', 8080, interval='10s')
)

# 健康检查失败时自动注销
# Consul会每10秒检查一次TCP连接

4.5.4 服务发现与负载均衡

客户端发现服务时，Consul会返回所有健康的实例：

# 查询服务
curl http://192.168.1.100:8500/v1/health/service/train-data-processor?passing=true

# 返回示例
[
  {
    "Node": "station-consul-01",
    "Service": {
      "ID": "train-data-processor-01",
      "Service": "train-data-processor",
      "Address": "192.168.1.50",
      "Port": 8080
    }
  },
  {
    "Node": "station-consul-02",
    "Service": {
      "ID": "train-data-processor-02",
      "Service": "train-data-processor",
      "Address": "192.168.1.51",
      "Port": 8080
    }
  }
]

客户端拿到列表后，自己做负载均衡。我一般用轮询或者最少连接数算法。嗯，这里要注意：如果某个实例挂了，Consul会在10秒内把它从列表里移除，客户端最多只会有10秒的“脏数据”。这个时间窗口在轨道交通里是可以接受的。

4.6 整体架构总结

好了，咱们把这一章的内容串起来。一个典型的ATS中间件架构是这样的：

• Kafka：负责海量数据（列车位置、设备日志）的采集和分发
• RabbitMQ：负责控制命令和告警的可靠投递
• DDS：负责实时性要求极高的数据（比如列车控制命令）
• Consul：负责所有服务的注册、发现和健康检查

说白了，就是让每种中间件干自己最擅长的事。别想着用一个东西解决所有问题——我见过有人只用Kafka做一切，结果命令延迟高得离谱，调度员都投诉了。

下一章咱们聊聊数据库的部署与调优，那是另一个有意思的话题。