本章深入探讨 ROS2 的核心通信机制，包括话题、服务、动作和参数服务器的实现原理。我们将从 DDS 层面剖析消息传递机制，理解 QoS 策略的应用场景，掌握高性能通信系统的设计模式。对于构建复杂机器人系统而言，选择合适的通信模式和优化策略至关重要。
ROS2 的话题通信基于 DDS（Data Distribution Service）标准实现，采用去中心化的发布-订阅模式：
Publisher Node Subscriber Nodes
[P1] ----\ /---> [S1]
\ /
[DDS Domain] ----+----> [S2]
/ \
[P2] ----/ \---> [S3]
核心特征：
ROS2 使用 IDL（Interface Definition Language）定义消息：
# geometry_msgs/msg/Twist.msg
Vector3 linear
float64 x
float64 y
float64 z
Vector3 angular
float64 x
float64 y
float64 z
消息序列化采用 CDR（Common Data Representation）标准，支持：
DDS 的 RTPS（Real-Time Publish-Subscribe）协议实现自动发现：
高性能场景下的内存优化策略：
传统模式：
App -> RMW -> DDS -> Socket -> Kernel -> Network
零拷贝模式（同进程）：
App -> Shared Memory -> App
零拷贝模式（跨进程）：
App -> POSIX SHM / Iceoryx -> App
实现零拷贝的条件：
服务采用请求-响应模式，适用于同步操作：
Client Server
|--- Request (call_id) -------->|
| | process()
|<------ Response (call_id) ----|
关键特性：
服务定义示例：
# AddTwoInts.srv
int64 a
int64 b
---
int64 sum
动作结合了话题和服务，支持长时间运行的任务：
Action Client Action Server
| |
|------ Goal Request ----------->|
|<----- Goal Response -----------|
| | execute()
|<----- Feedback (Topic) --------|
|<----- Feedback (Topic) --------|
| |
|------ Cancel Request --------->|
|<----- Cancel Response ---------|
| |
|<----- Result (Service) --------|
动作由五个服务和两个话题组成：
动作服务器维护目标状态机：
┌─────────┐
│ PENDING │
└────┬────┘
│ accept
┌────▼────┐
│EXECUTING│◄────┐
└────┬────┘ │
┌──┴──┐ ┌──┴────┐
cancel │ │ │PREEMPT│
▼ ▼ └───────┘
┌────────┐ ┌─────────┐
│CANCELED│ │SUCCEEDED│
└────────┘ └─────────┘
状态转换规则：
ROS2 采用分布式参数架构，每个节点维护自己的参数：
Node A Node B
┌──────────────┐ ┌──────────────┐
│ Parameters: │ │ Parameters: │
│ - param1 │ │ - param3 │
│ - param2 │ │ - param4 │
├──────────────┤ ├──────────────┤
│ Services: │ │ Services: │
│ - get │ │ - get │
│ - set │ │ - set │
│ - list │ │ - list │
└──────────────┘ └──────────────┘
参数类型支持：
参数变化通过事件话题广播：
/parameter_events (rcl_interfaces/msg/ParameterEvent)
├── node: string
├── new_parameters: Parameter[]
├── changed_parameters: Parameter[]
└── deleted_parameters: Parameter[]
监听参数变化的模式：
add_on_set_parameters_callback()/node_name/parameter_events/parameter_events参数描述符提供元数据：
parameter_descriptors:
velocity_limit:
type: DOUBLE
description: "Maximum velocity in m/s"
read_only: false
dynamic_typing: false
additional_constraints: "Must be positive"
floating_point_range:
- from_value: 0.0
to_value: 10.0
step: 0.1
| 特性 | 话题 | 服务 | 动作 | 参数 |
|---|---|---|---|---|
| 通信模式 | 异步多对多 | 同步一对一 | 异步+反馈 | 键值存储 |
| 数据流向 | 单向 | 双向 | 双向+状态 | 双向 |
| 适用频率 | 高频(>100Hz) | 低频(<10Hz) | 长任务 | 配置更新 |
| 缓冲支持 | 是 | 否 | 部分 | 否 |
| QoS配置 | 完整 | 有限 | 混合 | 无 |
| 典型延迟 | <1ms | 1-10ms | 可变 | 10-100ms |
需要通信？
├─ 是周期性数据流？
│ ├─ 是 → 使用话题
│ │ ├─ 需要可靠传输？→ RELIABLE QoS
│ │ └─ 容忍丢包？→ BEST_EFFORT QoS
│ └─ 否
│ ├─ 是请求-响应？
│ │ ├─ 需要快速响应？→ 服务
│ │ └─ 长时间任务？→ 动作
│ └─ 是配置数据？→ 参数
通信开销分析（基于 1KB 消息）：
话题（局域网）：
- 建立连接：~10ms (仅首次)
- 单次传输：~0.1-0.5ms
- 吞吐量：>10000 msg/s
服务（局域网）：
- 建立连接：~10ms
- 请求-响应：~1-5ms
- 吞吐量：~200 calls/s
动作（局域网）：
- 目标发送：~5ms
- 反馈更新：~0.5ms/次
- 结果获取：~5ms
ROS2 提供 10 个 QoS 策略维度：
ROS2 提供常用配置模板：
传感器数据（SensorDataQoS）：
- History: KEEP_LAST(5)
- Reliability: BEST_EFFORT
- Durability: VOLATILE
参数（ParameterQoS）：
- History: KEEP_LAST(1000)
- Reliability: RELIABLE
- Durability: VOLATILE
服务（ServicesQoS）：
- History: KEEP_LAST(10)
- Reliability: RELIABLE
- Durability: VOLATILE
系统默认（SystemDefaultQoS）：
- History: KEEP_LAST(10)
- Reliability: RELIABLE
- Durability: VOLATILE
发布者和订阅者 QoS 必须兼容：
| Policy | Publisher | Subscriber | Compatible |
|---|---|---|---|
| Reliability | BEST_EFFORT | BEST_EFFORT | ✓ |
| Reliability | BEST_EFFORT | RELIABLE | ✗ |
| Reliability | RELIABLE | BEST_EFFORT | ✓ |
| Reliability | RELIABLE | RELIABLE | ✓ |
| Durability | VOLATILE | VOLATILE | ✓ |
| Durability | VOLATILE | TRANSIENT_LOCAL | ✗ |
| Durability | TRANSIENT_LOCAL | VOLATILE | ✓ |
| 特性 | Cyclone DDS | Fast DDS | Connext DDS |
|---|---|---|---|
| 开源 | 是(Eclipse) | 是(eProsima) | 否(RTI) |
| 零拷贝 | iceoryx集成 | 原生支持 | 支持 |
| 安全扩展 | 基础 | 完整 | 完整 |
| 性能 | 低延迟优化 | 高吞吐优化 | 企业级 |
| 内存占用 | 小 | 中 | 大 |
| 实时性 | 良好 | 良好 | 优秀 |
Cyclone DDS 配置示例：
<CycloneDDS>
<Domain>
<General>
<NetworkInterfaceAddress>192.168.1.100</NetworkInterfaceAddress>
<MaxMessageSize>65500B</MaxMessageSize>
</General>
<Discovery>
<ParticipantIndex>0</ParticipantIndex>
<MaxAutoParticipantIndex>120</MaxAutoParticipantIndex>
<Multicast>
<Address>239.255.0.1</Address>
</Multicast>
</Discovery>
<Tracing>
<OutputFile>cyclone.log</OutputFile>
<Verbosity>warning</Verbosity>
</Tracing>
</Domain>
<DDSI2E>
<Internal>
<SocketReceiveBufferSize>2MiB</SocketReceiveBufferSize>
<SocketSendBufferSize>2MiB</SocketSendBufferSize>
<FragmentSize>4000B</FragmentSize>
</Internal>
</DDSI2E>
</CycloneDDS>
关键调优点：
# Linux 内核参数
sudo sysctl -w net.core.rmem_max=134217728
sudo sysctl -w net.core.wmem_max=134217728
sudo sysctl -w net.ipv4.udp_mem="102400 873800 16777216"
Optiver 是全球领先的做市商，其交易系统对延迟极其敏感。他们基于 ROS2 构建了实验性的硬件加速交易执行系统，用于期权定价和高频交易。
┌─────────────┐ Kernel Bypass ┌──────────────┐
│Market Data │◄────────────────────────►│FPGA Receiver │
│ Feed │ DPDK/RDMA │ (10G NIC) │
└─────────────┘ └──────┬───────┘
│ Zero Copy
▼
┌─────────────┐ Shared Memory ┌──────────────┐
│ Strategy │◄────────────────────────►│ ROS2 Node │
│ Engine │ iceoryx │ (RT Kernel) │
└─────────────┘ └──────┬───────┘
│
▼
┌─────────────┐ InfiniBand ┌──────────────┐
│Order Router │◄────────────────────────►│Exchange Gate │
│ │ RDMA │ │
└─────────────┘ └──────────────┘
// 使用 DPDK 绕过内核网络栈
rte_eth_rx_burst(port_id, queue_id, rx_pkts, MAX_BURST);
// 直接映射网卡队列到用户空间
void* rx_ring = mmap(NULL, ring_size,
PROT_READ | PROT_WRITE,
MAP_SHARED, nic_fd, 0);
// 隔离 CPU 核心
// /boot/cmdline: isolcpus=2-7 nohz_full=2-7 rcu_nocbs=2-7
cpu_set_t cpuset;
CPU_ZERO(&cpuset);
CPU_SET(isolated_core, &cpuset);
pthread_setaffinity_np(thread, sizeof(cpuset), &cpuset);
// 2MB 大页减少 TLB miss
void* pool = mmap(NULL, POOL_SIZE,
PROT_READ | PROT_WRITE,
MAP_PRIVATE | MAP_ANONYMOUS | MAP_HUGETLB,
-1, 0);
// 预热缓存
memset(pool, 0, POOL_SIZE);
<Cyclone>
<DDSI2E>
<General>
<CoalesceInterval>0ns</CoalesceInterval>
<MaxMessageSize>1400B</MaxMessageSize>
</General>
<Internal>
<ReceiveThreadMode>Exclusive</ReceiveThreadMode>
<ReceiveThreadPriority>99</ReceiveThreadPriority>
</Internal>
</DDSI2E>
</Cyclone>
实测结果（Intel Xeon Gold 6258R + Mellanox ConnectX-6）：
| 指标 | 原始 ROS2 | 优化后 | 提升 |
|---|---|---|---|
| 平均延迟 | 250μs | 8μs | 31x |
| P99 延迟 | 1.2ms | 45μs | 26x |
| P99.9 延迟 | 5ms | 120μs | 41x |
| 吞吐量 | 50K msg/s | 1.2M msg/s | 24x |
| CPU 使用率 | 45% | 98% | - |
| 内存带宽 | 2.1 GB/s | 28 GB/s | 13x |
问题：DDS 发现机制导致启动延迟 解决：使用静态发现，预配置所有端点
问题：垃圾回收导致延迟尖峰 解决：禁用动态内存分配，使用对象池
问题：中断导致抖动 解决：中断亲和性绑定到非关键核心
问题：时间戳不准确 解决：使用 TSC 时钟源，PTP 硬件时间戳
iceoryx 提供真正的零拷贝共享内存传输：
// 发布者端 - Loaned Message
auto loaned_msg = publisher->borrow_loaned_message();
loaned_msg.get().data = sensor_data; // 直接写入共享内存
publisher->publish(std::move(loaned_msg));
// 订阅者端 - Zero Copy
subscription->take_loaned_message(
[](const sensor_msgs::msg::PointCloud2& msg) {
// 直接访问共享内存，无拷贝
process_pointcloud(msg.data.data());
});
性能对比（10MB 点云数据）：
class TokenBucket {
std::chrono::steady_clock::time_point last_refill_;
double tokens_;
double capacity_;
double refill_rate_;
public:
bool try_consume(size_t bytes) {
refill();
double required = bytes / 1024.0; // KB
if (tokens_ >= required) {
tokens_ -= required;
return true;
}
return false;
}
void refill() {
auto now = std::chrono::steady_clock::now();
auto elapsed = now - last_refill_;
double new_tokens = elapsed.count() * refill_rate_;
tokens_ = std::min(capacity_, tokens_ + new_tokens);
last_refill_ = now;
}
};
基于网络状况动态调整 QoS：
class AdaptiveQoS {
void adjust_qos_based_on_rtt(double rtt_ms) {
if (rtt_ms > 100) {
// 网络拥塞，降低发送频率
qos_.reliability(RMW_QOS_POLICY_RELIABILITY_BEST_EFFORT);
qos_.history(RMW_QOS_POLICY_HISTORY_KEEP_LAST);
qos_.depth(1);
} else if (rtt_ms < 10) {
// 网络良好，提高可靠性
qos_.reliability(RMW_QOS_POLICY_RELIABILITY_RELIABLE);
qos_.depth(10);
}
}
};
// 跨节点追踪上下文传播
class TracedPublisher {
void publish_with_trace(const Message& msg) {
auto span = tracer->StartSpan("publish_message");
// 注入追踪上下文到消息头
TextMapCarrier carrier;
propagator->Inject(carrier, span->GetContext());
msg._metadata.trace_parent = carrier.Get("traceparent");
msg._metadata.trace_state = carrier.Get("tracestate");
publisher_->publish(msg);
span->End();
}
};
生成的追踪数据可视化：
[Node A: Sensor] [Node B: Filter] [Node C: Control]
│ │ │
├──publish(100ms)──────────►│ │
│ ├──process(50ms)───► │
│ │ └──►compute(30ms)
│ │ │
└───────────────────────────┴────────────────────────┘
Total Latency: 180ms
应用层：Topics, Services, Actions, Parameters
↓
中间件层：rmw (ROS Middleware)
↓
DDS层：RTPS Discovery, QoS Policies
↓
传输层：UDP/TCP, Shared Memory
性能优化公式
总延迟 = 序列化时间 + 传输时间 + 反序列化时间 + 处理时间
其中：
QoS 匹配规则
兼容性 = ∏(Policy_pub ⊇ Policy_sub)
其中 ⊇ 表示发布者策略满足订阅者要求
症状：发布者和订阅者都正常运行，但收不到数据
诊断：
ros2 topic info /topic_name --verbose
# 检查 QoS 配置是否匹配
解决：使用 QoS override 或统一配置
// 自适应 QoS
rmw_qos_profile_t qos = rmw_qos_profile_sensor_data;
if (need_reliable) {
qos.reliability = RMW_QOS_POLICY_RELIABILITY_RELIABLE;
}
症状：小消息正常，大于 64KB 消息丢失
原因：UDP 分片在拥塞时易丢失
解决：
症状：服务调用挂起，节点无响应
代码示例（错误）：
void callback() {
auto future = client->async_send_request(request);
future.wait(); // 死锁！回调中阻塞
}
解决：使用异步模式或独立执行器
症状：参数设置后立即读取得到旧值
原因：参数服务异步处理
解决：使用参数事件确认更新
auto result = node->set_parameter(param);
if (result.successful) {
// 等待参数事件确认
wait_for_parameter_event(param_name);
}
症状：新目标取消了正在执行的目标
原因：默认 GoalID 重用
解决：
// 使用唯一 GoalID
goal_msg.goal_id.uuid = generate_uuid();
# 监控话题
ros2 topic echo /topic --qos-profile sensor_data
# 检查节点图
ros2 node info /node_name
# DDS 调试
export RMW_IMPLEMENTATION=rmw_cyclonedds_cpp
export CYCLONEDDS_TRACE=trace.log
# CPU 分析
perf record -g ros2 run package node
perf report
# 网络分析
tcpdump -i lo -w ros2.pcap 'udp port 7400'
# 追踪分析
ros2 trace start session_name
# 内存泄漏
valgrind --leak-check=full ros2 run package node
# 内存分析
heaptrack ros2 run package node
练习 6.1：话题通信延迟分析
给定一个 ROS2 系统，发布者以 100Hz 发送 1MB 的点云数据，网络带宽为 1Gbps，计算：
💡 提示：考虑序列化、传输、反序列化三个阶段
练习 6.2：QoS 兼容性判断
判断以下发布者和订阅者配置是否兼容，并说明原因：
| 场景 | Publisher QoS | Subscriber QoS | 兼容? |
|---|---|---|---|
| A | RELIABLE, TRANSIENT_LOCAL | BEST_EFFORT, VOLATILE | ? |
| B | BEST_EFFORT, VOLATILE | RELIABLE, VOLATILE | ? |
| C | RELIABLE, VOLATILE | RELIABLE, TRANSIENT_LOCAL | ? |
💡 提示：发布者必须满足订阅者的最低要求
练习 6.3：服务超时设计
设计一个服务调用，要求：
💡 提示：使用 future 和 chrono