本文由 linuxROS 整理发布，首发于 linuxros.cn，转载请注明出处。

ROS2多机器人协作与AI集成实战：从DDS隔离到智能推理管线

导读：多机器人系统如何互不干扰？MoveIt2怎么配置运动规划？YOLOv8如何接入ROS2话题？本篇覆盖多机器人协作、MoveIt2工业机器人、AI推理集成、DDS安全与网络配置，代码可直接运行。

一、多机器人系统协议

DDS Domain ID隔离

ROS2基于DDS通信，Domain ID是最基础的隔离手段。不同Domain ID的机器人完全隔离，彼此看不到对方的话题和服务。

# 终端1：Robot1 使用 Domain 10
export ROS_DOMAIN_ID=10
ros2 run demo_nodes_cpp talker

# 终端2：Robot2 使用 Domain 20
export ROS_DOMAIN_ID=20
ros2 run demo_nodes_cpp listener
# 不会收到Robot1的消息，因为Domain不同

Domain ID取值范围0~232（Fast DDS），0~101（Cyclone DDS）。建议在~/.bashrc中固定设置：

# 写入bashrc，避免每次手动设置
echo "export ROS_DOMAIN_ID=10" >> ~/.bashrc
source ~/.bashrc

命名空间隔离

同一Domain内，用namespace区分不同机器人。同一话题名在不同namespace下互不冲突：

# Robot1：namespace=/robot1
ros2 run demo_nodes_cpp talker --ros-args --remap __ns:=/robot1

# Robot2：namespace=/robot2
ros2 run demo_nodes_cpp talker --ros-args --remap __ns:=/robot2

# 查看话题列表
ros2 topic list
# /robot1/chatter
# /robot2/chatter

Launch文件配置多机器人

用Python Launch文件批量启动多机器人，通过namespace参数隔离：

# multi_robot_launch.py
from launch import LaunchDescription
from launch_ros.actions import Node

def generate_launch_description():
    robots = []
    for i in range(3):
        robot = Node(
            package='demo_nodes_cpp',
            executable='talker',
            namespace=f'robot{i}',
            name=f'talker_node_{i}',
            output='screen',
        )
        robots.append(robot)

    return LaunchDescription(robots)

运行：

ros2 launch multi_robot_launch.py

多机器人通信架构

多机器人协调策略

策略	架构	优点	缺点	适用场景
集中式	中心节点调度	逻辑清晰、易调试	单点故障、扩展性差	3~5台机器人
分布式	机器人自主协商	容错性强、可扩展	一致性难保证	大规模集群
混合法	分层：组内集中组间分布	兼顾性能与容错	实现复杂	10台以上

集中式示例——中央调度节点：

# central_scheduler.py
import rclpy
from rclpy.node import Node
from std_msgs.msg import String

class CentralScheduler(Node):
    def __init__(self):
        super().__init__('central_scheduler')
        self.robot_count = 3
        # 订阅各机器人状态
        self.status_subs = []
        for i in range(self.robot_count):
            sub = self.create_subscription(
                String, f'/robot{i}/status',
                lambda msg, rid=i: self.on_status(rid, msg), 10)
            self.status_subs.append(sub)
        # 发布任务指令
        self.cmd_pubs = []
        for i in range(self.robot_count):
            pub = self.create_publisher(String, f'/robot{i}/cmd', 10)
            self.cmd_pubs.append(pub)
        self.robot_status = {}

    def on_status(self, robot_id, msg):
        self.robot_status[robot_id] = msg.data
        self.get_logger().info(f'Robot{robot_id}: {msg.data}')
        # 简单调度：空闲机器人分配任务
        if msg.data == 'idle':
            cmd = String()
            cmd.data = 'goto:waypoint_A'
            self.cmd_pubs[robot_id].publish(cmd)

def main():
    rclpy.init()
    node = CentralScheduler()
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()

二、MoveIt2工业机器人

MoveIt2核心组件

MoveIt2是ROS2 Jazzy下的运动规划框架，核心组件：

组件	职责
MoveGroupInterface	用户API入口，封装规划执行
PlanningPipeline	运动规划管线，调用规划器生成轨迹
CollisionChecking	碰撞检测，基于FCL库
TrajectoryExecution	轨迹执行，对接ros2_control

MoveIt2架构

flowchart TB API["用户API MoveGroupInterface"] --> MG["MoveGroup 规划协调"] MG --> PP["PlanningPipeline OMPL/Pilz/CHOMP"] PP --> CC["CollisionCheck FCL碰撞检测"] CC -->|"无碰撞"| TE["TrajectoryExecution 轨迹执行"] CC -->|"有碰撞"| PP TE --> RC["ros2_control 硬件接口"] RC --> HW["机器人硬件"] style API fill:#E8F5E9 style MG fill:#E3F2FD style PP fill:#FFF8E1 style CC fill:#FFEBEE style TE fill:#E3F2FD style RC fill:#F3E5F5 style HW fill:#E8F5E9

URDF + SRDF建模

URDF定义机器人几何和运动学结构，SRDF（Semantic Robot Description）定义语义信息——规划组、默认姿态、碰撞白名单：
URDF关键片段（6轴机械臂示例）：

<!-- urdf/robot.urdf.xacro -->
<robot xmlns:xacro="http://www.ros.org/wiki/xacro" name="my_robot">
  <xacro:include filename="joint_definitions.xacro"/>

  <!-- 基座 -->
  <link name="base_link">
    <visual>
      <geometry><cylinder radius="0.1" length="0.05"/></geometry>
    </visual>
  </link>

  <!-- 关节1 -->
  <joint name="joint1" type="revolute">
    <parent link="base_link"/>
    <child link="link1"/>
    <origin xyz="0 0 0.05" rpy="0 0 0"/>
    <axis xyz="0 0 1"/>
    <limit lower="-3.14" upper="3.14" effort="50" velocity="1.0"/>
  </joint>

  <link name="link1">
    <visual>
      <geometry><cylinder radius="0.04" length="0.3"/></geometry>
    </visual>
  </link>
</robot>

SRDF关键片段：

<!-- srdf/robot.srdf -->
<robot name="my_robot">
  <!-- 规划组：包含所有关节 -->
  <group name="manipulator">
    <chain base_link="base_link" tip_link="tool0"/>
  </group>

  <!-- 默认姿态 -->
  <group_state name="home" group="manipulator">
    <joint name="joint1" value="0"/>
    <joint name="joint2" value="0"/>
    <joint name="joint3" value="0"/>
  </group_state>

  <!-- 碰撞白名单：永远不检测的link对 -->
  <disable_collisions link1="base_link" link2="link1" reason="Adjacent"/>
</robot>

MoveIt2 Setup Assistant

# 启动配置助手
ros2 launch moveit_setup_assistant setup_assistant.launch.py

# 操作步骤：
# 1. 加载URDF
# 2. 生成SRDF（选择规划组、默认姿态）
# 3. 配置碰撞检测（采样数默认10万）
# 4. 配置控制器（ros2_control）
# 5. 生成配置包

Python运动规划示例

# moveit2_plan_example.py
import rclpy
from rclpy.node import Node
from moveit.planning import MoveItPy
from moveit.core.robot_state import RobotState

def main():
    rclpy.init()
    # 初始化MoveIt2
    moveit = MoveItPy(node_name="moveit_py")
    arm = moveit.get_planning_component("manipulator")

    # 设置起始姿态为当前状态
    arm.set_start_state_to_current_state()

    # 设置目标位姿
    arm.set_goal_state(configuration_name="home")

    # 规划
    plan_result = arm.plan()
    if plan_result:
        # 执行
        moveit.execute(plan_result.trajectory)
        print("执行成功")
    else:
        print("规划失败")

    rclpy.shutdown()

if __name__ == "__main__":
    main()

使用MoveGroupInterface的Python示例（更常用）：

# move_group_example.py
import rclpy
from rclpy.node import Node
from moveit_msgs.msg import MoveGroupAction, MoveGroupGoal
from geometry_msgs.msg import PoseStamped

class MoveGroupClient(Node):
    def __init__(self):
        super().__init__('move_group_client')
        self.get_logger().info('MoveGroup客户端已启动')

    def plan_to_pose(self, target_pose):
        """规划到目标位姿"""
        # 通过MoveIt2的Python API进行规划
        # 实际项目中使用 moveit_py 接口
        self.get_logger().info(f'规划到目标 {target_pose}')

def main():
    rclpy.init()
    node = MoveGroupClient()
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()

常用规划器对比

规划器	算法	特点	适用场景
OMPL	RRT/PRM/BIT*	采样型，概率完备	6轴以上机械臂
Pilz	LIN/CIRC	确定性，笛卡尔直线/圆弧	焊接、涂胶
CHOMP	梯度优化	优化型，轨迹平滑	避障精细操作
STOMP	随机优化	不需梯度，鲁棒	高维空间

三、ROS2与AI/ML集成

AI推理管线架构

传感器数据经过预处理、推理、后处理，最终发布到ROS2话题。

flowchart TB Sensor["传感器 Camera/LiDAR"] --> Pre["预处理 Resize/Norm"] Pre --> Engine["推理引擎 YOLOv8/TensorRT"] Engine --> Post["后处理 NMS/坐标变换"] Post --> Topic["ROS2话题 Detection2D"] Topic --> Nav["导航/抓取 下游节点"] style Sensor fill:#E3F2FD style Pre fill:#FFF8E1 style Engine fill:#F3E5F5 style Post fill:#FFF8E1 style Topic fill:#E8F5E9 style Nav fill:#E3F2FD

YOLOv8目标检测集成

安装依赖：

pip install ultralytics opencv-python sensor-msgs

完整节点代码——订阅Image话题，YOLOv8推理，发布检测结果：

# yolo_detector_node.py
import rclpy
from rclpy.node import Node
from sensor_msgs.msg import Image
from vision_msgs.msg import Detection2DArray, Detection2D, ObjectHypothesisWithPose
from cv_bridge import CvBridge
from ultralytics import YOLO
import numpy as np

class YoloDetectorNode(Node):
    def __init__(self):
        super().__init__('yolo_detector')

        # 加载模型
        self.declare_parameter('model_path', 'yolov8n.pt')
        model_path = self.get_parameter('model_path').value
        self.model = YOLO(model_path)
        self.get_logger().info(f'模型加载完成: {model_path}')

        self.bridge = CvBridge()
        self.conf_threshold = 0.5

        # 订阅图像话题
        self.sub = self.create_subscription(
            Image, '/camera/image_raw', self.image_callback, 10)

        # 发布检测结果
        self.det_pub = self.create_publisher(
            Detection2DArray, '/detections', 10)

        # 发布标注图像（调试用）
        self.annotated_pub = self.create_publisher(
            Image, '/camera/annotated', 10)

    def image_callback(self, msg):
        # ROS Image → OpenCV
        cv_img = self.bridge.imgmsg_to_cv2(msg, desired_encoding='bgr8')

        # YOLOv8推理
        results = self.model(cv_img, conf=self.conf_threshold, verbose=False)

        # 构造检测结果消息
        det_array = Detection2DArray()
        det_array.header = msg.header

        for result in results:
            boxes = result.boxes
            for box in boxes:
                det = Detection2D()
                det.bbox.center.x = float(box.xywh[0][0])
                det.bbox.center.y = float(box.xywh[0][1])
                det.bbox.size_x = float(box.xywh[0][2])
                det.bbox.size_y = float(box.xywh[0][3])

                hypothesis = ObjectHypothesisWithPose()
                hypothesis.hypothesis.class_id = str(int(box.cls[0]))
                hypothesis.hypothesis.score = float(box.conf[0])
                det.results.append(hypothesis)

                det_array.detections.append(det)

        self.det_pub.publish(det_array)

        # 发布标注图像
        annotated = results[0].plot()
        annotated_msg = self.bridge.cv2_to_imgmsg(annotated, encoding='bgr8')
        self.annotated_pub.publish(annotated_msg)

def main():
    rclpy.init()
    node = YoloDetectorNode()
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()

if __name__ == "__main__":
    main()

运行：

# 启动检测节点
ros2 run my_ai_pkg yolo_detector_node --ros-args \
  -p model_path:=yolov8n.pt

# 查看检测结果
ros2 topic echo /detections

TensorRT加速推理

# yolo_trt_node.py（TensorRT加速版本）
import rclpy
from rclpy.node import Node
from sensor_msgs.msg import Image
from vision_msgs.msg import Detection2DArray, Detection2D, ObjectHypothesisWithPose
from cv_bridge import CvBridge
from ultralytics import YOLO

class YoloTrtNode(Node):
    def __init__(self):
        super().__init__('yolo_trt_detector')

        # 导出并加载TensorRT引擎
        self.declare_parameter('model_path', 'yolov8n.pt')
        model_path = self.get_parameter('model_path').value

        # 首次运行自动导出engine，后续直接加载
        self.model = YOLO(model_path, task='detect')

        self.bridge = CvBridge()
        self.sub = self.create_subscription(
            Image, '/camera/image_raw', self.image_callback, 10)
        self.det_pub = self.create_publisher(
            Detection2DArray, '/detections', 10)

        self.get_logger().info('TensorRT检测节点已启动')

    def image_callback(self, msg):
        cv_img = self.bridge.imgmsg_to_cv2(msg, desired_encoding='bgr8')

        # 使用TensorRT引擎推理
        results = self.model(cv_img, conf=0.5, verbose=False)

        det_array = Detection2DArray()
        det_array.header = msg.header

        for result in results:
            for box in result.boxes:
                det = Detection2D()
                det.bbox.center.x = float(box.xywh[0][0])
                det.bbox.center.y = float(box.xywh[0][1])
                det.bbox.size_x = float(box.xywh[0][2])
                det.bbox.size_y = float(box.xywh[0][3])

                hypothesis = ObjectHypothesisWithPose()
                hypothesis.hypothesis.class_id = str(int(box.cls[0]))
                hypothesis.hypothesis.score = float(box.conf[0])
                det.results.append(hypothesis)
                det_array.detections.append(det)

        self.det_pub.publish(det_array)

def main():
    rclpy.init()
    node = YoloTrtNode()
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()

if __name__ == "__main__":
    main()

导出TensorRT引擎：

# 导出为TensorRT engine（FP16精度量化）
yolo export model=yolov8n.pt format=engine half=True

# 生成的engine文件在同级目录：yolov8n.engine

边缘端 vs 云端部署

维度	边缘端（Jetson）	云端（GPU服务器）
延迟	低（<10ms）	高（网络+推理50~200ms）
带宽	无需传输图像	需上传图像数据
算力	有限（Jetson Orin 275 TOPS）	充足（A100 312 TOPS）
可靠性	不依赖网络	依赖网络连接
适用	实时避障/抓取	批量分析/训练

四、ROS2安全机制

DDS-Security机制

ROS2 Jazzy的DDS实现（Fast DDS / Cyclone DDS）支持DDS-Security规范，提供以下安全能力：

来自 linuxros.cn · linuxROS

能力	说明	插件
认证	验证节点身份	Authentication
加密	传输数据加密	Cryptographic
访问控制	话题/服务权限控制	Access Control
日志	安全事件审计	Logging
数据标记	防篡改标注	Data Tagging

SROS2配置工具

SROS2是ROS2官方安全配置工具，自动生成证书和策略文件。

# 创建安全目录
mkdir -p ~/sros2_keystore

# 初始化密钥库（生成CA证书）
ros2 security create_keystore ~/sros2_keystore

# 为节点签发证书
ros2 security create_enclave ~/sros2_keystore /my_node
ros2 security create_enclave ~/sros2_keystore /talker_node
ros2 security create_enclave ~/sros2_keystore /listener_node

# 查看生成的文件
ls ~/sros2_keystore/
# ca.cert.pem  ca.key.pem  ca.governance.pem
# my_node/  talker_node/  listener_node/

安全策略配置

策略文件（policy.xml）定义节点的发布/订阅权限：

<!-- policy.xml -->
<dds xmlns="http://www.omg.org/dds" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
  <permissions>
    <grant name="/talker_node">
      <subject_name>CN=/talker_node</subject_name>
      <validity>
        <not_before>2024-01-01T00:00:00</not_before>
        <not_after>2030-01-01T00:00:00</not_after>
      </validity>
      <allow_rule>
        <domains><id>10</id></domains>
        <publish>
          <topics><topic>/chatter</topic></topics>
        </publish>
      </allow_rule>
      <default>DENY</default>
    </grant>

    <grant name="/listener_node">
      <subject_name>CN=/listener_node</subject_name>
      <validity>
        <not_before>2024-01-01T00:00:00</not_before>
        <not_after>2030-01-01T00:00:00</not_after>
      </validity>
      <allow_rule>
        <domains><id>10</id></domains>
        <subscribe>
          <topics><topic>/chatter</topic></topics>
        </subscribe>
      </allow_rule>
      <default>DENY</default>
    </grant>
  </permissions>
</dds>

启用安全模式

# 设置安全目录环境变量
export ROS_SECURITY_KEYSTORE=~/sros2_keystore
export ROS_SECURITY_ENABLE=true
export ROS_SECURITY_STRATEGY=Enforce  # Enforce强制 / Permissive宽松

# 启动节点时指定enclave
ros2 run demo_nodes_cpp talker --ros-args --enclave /talker_node

# 另一个终端
ros2 run demo_nodes_cpp listener --ros-args --enclave /listener_node

`Enforce`模式下，没有有效证书的节点无法加入通信。`Permissive`模式下，安全验证失败的节点仍可降级运行。

五、网络配置

DDS发现机制

DDS节点通过发现机制找到彼此，支持两种模式：

模式	原理	适用场景
多播（Multicast）	组播地址广播发现	同一局域网
单播（Unicast）	指定IP地址发现	跨网络/VPN

默认使用多播，局域网内即插即用。跨网段需配置单播。

Fast DDS配置文件

<!-- fastdds_profile.xml -->
<?xml version="1.0" encoding="UTF-8"?>
<dds>
  <profiles xmlns="http://www.eprosima.com/XMLSchemas/fastRTPS_Profiles">

    <!-- 默认配置：局域网多播 -->
    <transport_descriptors>
      <transport_descriptor>
        <transport_id>udp_transport</transport_id>
        <type>UDPv4</type>
      </transport_descriptor>
    </transport_descriptors>

    <!-- 跨网段单播配置 -->
    <participant profile_name="cross_network" is_default_profile="true">
      <rtps>
        <useBuiltinTransports>false</useBuiltinTransports>
        <userTransports>
          <transport_id>udp_transport</transport_id>
        </userTransports>
        <builtinTransports max_msg_size="65536"/>
        <initialPeersList>
          <locator>
            <udpv4>
              <!-- 对端机器人IP -->
              <address>192.168.2.100</address>
              <port>7400</port>
            </udpv4>
          </locator>
        </initialPeersList>
      </rtps>
    </participant>

  </profiles>
</dds>

使用配置文件：

export FASTRTPS_DEFAULT_PROFILES_FILE=/path/to/fastdds_profile.xml
export ROS_DOMAIN_ID=10

# 启动节点，自动加载配置
ros2 run demo_nodes_cpp talker

网络延迟优化

优化手段	配置	效果
共享内存传输	Fast DDS SHM	同机通信延迟降低80%
QoS可靠性降级	RELIABLE → BEST_EFFORT	减少重传，降低延迟
数据类型精简	去掉冗余字段	减少序列化开销
历史深度控制	depth=1（只保留最新）	减少内存占用
UDP缓冲区调优	sysctl net.core.rmem_max	避免丢包

共享内存传输配置：

<!-- fastdds_shm.xml -->
<dds>
  <profiles xmlns="http://www.eprosima.com/XMLSchemas/fastRTPS_Profiles">
    <transport_descriptors>
      <transport_descriptor>
        <transport_id>shm_transport</transport_id>
        <type>SHM</type>
        <max_msg_size>65536</max_msg_size>
        <segment_size>524288</segment_size>
      </transport_descriptor>
    </transport_descriptors>

    <participant profile_name="shm_profile" is_default_profile="true">
      <rtps>
        <useBuiltinTransports>false</useBuiltinTransports>
        <userTransports>
          <transport_id>shm_transport</transport_id>
        </userTransports>
      </rtps>
    </participant>
  </profiles>
</dds>

ROS2 over VPN/WAN

跨网络通信需要解决两个问题：多播不可达、NAT穿透。

# WireGuard VPN配置示例
# 机器人端：192.168.170.128
[Interface]
PrivateKey = <robot_private_key>
Address = 10.0.0.2/24

[Peer]
PublicKey = <server_public_key>
Endpoint = <server_ip>:51820
AllowedIPs = 10.0.0.0/24
PersistentKeepalive = 25

# 服务器端
[Interface]
PrivateKey = <server_private_key>
Address = 10.0.0.1/24
ListenPort = 51820

[Peer]
PublicKey = <robot_public_key>
AllowedIPs = 10.0.0.2/32

VPN连通后，配置Fast DDS单播发现，使用VPN隧道IP即可跨网络通信。

六、常见问题

Q1：多机器人话题冲突怎么办？

命名空间隔离。在Launch文件中为每个机器人设置不同的namespace，话题自动加上前缀（如/robot1/cmd_vel）。如果需要跨机器人通信，确保订阅方使用完整的带namespace话题名。

Q2：MoveIt2规划失败怎么排查？

三步排查：
1. 检查URDF关节限制——<limit>的lower/upper是否合理
2. 检查碰撞检测——用RViz2的MotionPlanning插件可视化碰撞点
3. 检查目标位姿是否在工作空间内——用moveit_py的get_current_state()对比

# RViz2中启动MoveIt2可视化
ros2 launch my_robot_moveit_config moveit_rviz.launch.py

Q3：AI推理延迟高怎么优化？

优先级排序：
1. 导出TensorRT引擎，FP16精度——通常提速~5倍
2. 降低输入分辨率——640×480代替1280×720
3. 使用更小的模型——YOLOv8n代替YOLOv8x
4. 开启异步推理——推理和回调分离，避免阻塞

# 异步推理示例
import threading

class AsyncDetector:
    def __init__(self):
        self.latest_frame = None
        self.lock = threading.Lock()

    def image_callback(self, msg):
        with self.lock:
            self.latest_frame = msg

    def inference_loop(self):
        while rclpy.ok():
            with self.lock:
                frame = self.latest_frame
            if frame is not None:
                results = self.model(frame)
                self.publish_results(results)

Q4：DDS跨网络不通怎么排查？

逐步排查：
1. 确认网络连通——ping对端IP
2. 确认Domain ID一致——echo $ROS_DOMAIN_ID
3. 检查防火墙——DDS默认端口7400~7404
4. 多播不可达时切换单播——配置Fast DDS initialPeersList
5. 检查NAT——VPN隧道内通信避免NAT问题

# 检查防火墙（Ubuntu）
sudo ufw status
sudo ufw allow 7400:7404/udp

# 检查DDS发现
ros2 daemon stop && ros2 daemon start
ros2 node list

七、总结

ROS2多机器人协作到AI集成的链路：

多机器人：Domain ID隔离不同团队，namespace隔离同组机器人，Launch文件统一管理
MoveIt2：URDF+SRDF建模，Setup Assistant生成配置，OMPL/Pilz/CHOMP按场景选规划器
AI集成：YOLOv8接入ROS2话题，TensorRT加速推理，边缘端/云端按延迟需求选择
安全：SROS2生成证书，policy.xml控制权限，Enforce模式强制安全
网络：局域网多播即插即用，跨网段单播配置，VPN穿透远程通信

速查表：

场景	方案
多机器人隔离	Domain ID + namespace
机械臂运动规划	MoveIt2 + OMPL
笛卡尔直线运动	MoveIt2 + Pilz
目标检测集成	YOLOv8 + vision_msgs
推理加速	TensorRT FP16
节点认证加密	SROS2 + DDS-Security
跨网段通信	Fast DDS单播 + initialPeersList
远程通信	WireGuard VPN + DDS单播
同机低延迟	Fast DDS共享内存

本文首发于linuxros.cn，转载请注明出处。