介绍

主题是 ros2 提供的三种主要接口方式之一，它通常被用于连续的数据流，如传感器数据、机器人状态等。
ros2 是一个强类型的匿名发布/订阅系统，它将复杂系统分解为许多模块化的节点，其中一些节点可以作为发布者、另外一些节点作为订阅者。主题是 ros中的重要元素，用于节点交换消息的总线。 一个节点可以发布数据到多个topic中，也可以从多个topic中订阅数据。

1发布者 - 1订阅者	m发布者 - n订阅者
,

本文基于python，从零开始创建一个基于发布-订阅的ros主题通信案例。

创建步骤

体验官方 talker listener 案例

我们先来看看官方的案例，运行一个talker和一个listener，其中talker每秒发布一条 Hello World: n 的消息到 话题chatter中，listener订阅该话题并输出。

启动 talker
$ ros2 run demo_nodes_cpp talker 或者 demo_nodes_py talker
启动 listener
$ ros2 run demo_nodes_py listener

可以在 /opt/ros/jazzy/lib/python3.12/site-packages/demo_nodes_py/topics 查看官方案例源码

控制台输出

rqt_graph 查看节点和topic关系

基于python开发发布订阅案例

1. 创建包 py_pubsub

   $ cd dev_ws/src
   $ ros2 pkg create --build-type ament_python --license Apache-2.0 py_pubsub
   

2. 创建 publisher 节点
   vim publisher_member_function.py

   import rclpy
   from rclpy.node import Node
   
   from std_msgs.msg import String
   
   
   class MinimalPublisher(Node):
   
       def __init__(self):
           super().__init__('minimal_publisher')
           self.publisher_ = self.create_publisher(String, 'topic', 10)
           timer_period = 0.5  # seconds
           self.timer = self.create_timer(timer_period, self.timer_callback)
           self.i = 0
   
       def timer_callback(self):
           msg = String()
           msg.data = 'Hello World: %d' % self.i
           self.publisher_.publish(msg)
           self.get_logger().info('Publishing: "%s"' % msg.data)
           self.i += 1
   
   
   def main(args=None):
       rclpy.init(args=args)
       minimal_publisher = MinimalPublisher()
       rclpy.spin(minimal_publisher)
       minimal_publisher.destroy_node()
       rclpy.shutdown()
   
   
   if __name__ == '__main__':
       main()
   

3. 创建 subscriber 节点
   vim subscriber_member_function.py

   import rclpy
   from rclpy.node import Node
   
   from std_msgs.msg import String
   
   
   class MinimalSubscriber(Node):
   
       def __init__(self):
           super().__init__('minimal_subscriber')
           self.subscription = self.create_subscription(
               String,
               'topic',
               self.listener_callback,
               10)
           self.subscription  # prevent unused variable warning
   
       def listener_callback(self, msg):
           self.get_logger().info('I heard: "%s"' % msg.data)
   
   
   def main(args=None):
       rclpy.init(args=args)
       minimal_subscriber = MinimalSubscriber()
       rclpy.spin(minimal_subscriber)
       minimal_subscriber.destroy_node()
       rclpy.shutdown()
   
   
   if __name__ == '__main__':
       main()
   

4. 在 package.xml 中新增依赖项

     <exec_depend>rclpy</exec_depend>
     <exec_depend>std_msgs</exec_depend>
   

5. 在 setup.py中新增 entry point

   entry_points={
           'console_scripts': [
                   'talker = py_pubsub.publisher_member_function:main',
                   'listener = py_pubsub.subscriber_member_function:main',
           ],
   },
   

6. 构建 & 运行

   构建
   $ cd dev_ws
   $ colcon build --packages-select py_pubsub
   运行
   $ source install/setup.bash
   $ ros2 run py_pubsub talker
   $ ros2 run py_pubsub listener
   

   控制台输出
   rqt_graph 查看节点和topic关系
   

注意事项

待添加

说明

软件版本
ubuntu24.04 Desktop
ros2 jazzy
python 3.12.4(conda)
参考文档
ros2 jazzy 官方文档 Writing a simple publisher and subscriber (Python)
ros2 jazzy 官方文档 Ubuntu-Development-Setup.html#talker-listener
ROS2学习笔记三：话题Topic