ROS2话题通信Python实现详解

Zorion2

13人浏览 · 2026-07-06 22:23:55

Zorion2 · 2026-07-06 22:23:55 发布

3.1 编写发布方代码（publisher.py）

3.2 编写订阅方代码（subscriber.py）

4.3 编写发布方代码（publisher_custom.py）

4.4 编写订阅方代码（subscriber_custom.py）

4.5 运行节点

五、总结

一、引言

在机器人操作系统ROS2中，话题通信是实现节点间数据传输的重要方式之一。通过话题通信，一个节点可以发布数据，而其他节点可以订阅这些数据，从而实现信息的共享和交互。本文将详细介绍如何使用Python代码在ROS2中实现话题通信，并通过实际案例进行演示。

二、准备工作

2.1 安装ROS2

确保已经在系统中正确安装了ROS2。以Ubuntu系统为例，可以按照官方文档的步骤进行安装。

2.2 创建工作空间和功能包

打开终端，进入想要创建工作空间的目录，然后执行以下命令创建工作空间和功能包：

mkdir -p ~/ros2_ws/src
cd ~/ros2_ws/src
ros2 pkg create --build-type ament_python py_topic_demo
cd..
colcon build

这将创建一个名为ros2_ws的工作空间，并在其中创建一个名为py_topic_demo的Python功能包。

三、简单文本话题通信

3.1 编写发布方代码（publisher.py）

在py_topic_demo功能包的src目录下创建publisher.py文件，内容如下：

import rclpy
from rclpy.node import Node
from std_msgs.msg import String

class SimplePublisher(Node):
    def __init__(self):
        super().__init__('simple_publisher')
        self.publisher_ = self.create_publisher(String, 'topic', 10)
        timer_period = 1  # 发布频率为1Hz
        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('发布的消息: "%s"' % msg.data)
        self.i += 1

def main(args=None):
    rclpy.init(args=args)
    simple_publisher = SimplePublisher()
    rclpy.spin(simple_publisher)
    simple_publisher.destroy_node()
    rclpy.shutdown()

3.2 编写订阅方代码（subscriber.py）

在同一src目录下创建subscriber.py文件，内容如下：

import rclpy
from rclpy.node import Node
from std_msgs.msg import String

class SimpleSubscriber(Node):
    def __init__(self):
        super().__init__('simple_subscriber')
        self.subscription = self.create_subscription(
            String,
            'topic',
            self.listener_callback,
            10)
        self.subscription  # 防止未使用变量警告

    def listener_callback(self, msg):
        self.get_logger().info('订阅的消息: "%s"' % msg.data)

def main(args=None):
    rclpy.init(args=args)
    simple_subscriber = SimpleSubscriber()
    rclpy.spin(simple_subscriber)
    simple_subscriber.destroy_node()
    rclpy.shutdown()

3.3 运行节点

打开两个终端，分别执行以下命令：

终端1（运行发布方）：

source ~/ros2_ws/install/setup.bash
ros2 run py_topic_demo publisher.py

终端2（运行订阅方）：

source ~/ros2_ws/install/setup.bash
ros2 run py_topic_demo subscriber.py

此时，发布方将以1Hz的频率发布文本消息，订阅方将接收到这些消息并打印到终端。

四、自定义接口话题通信

4.1 定义自定义接口消息

在py_topic_demo功能包下创建msg目录，然后在其中创建Student.msg文件，内容如下：

string name
int32 age
float32 height

4.2 配置功能包

编辑py_topic_demo功能包下的package.xml文件，添加以下依赖：

<build_depend>message_generation</build_depend>
<exec_depend>message_runtime</exec_depend>

同时，编辑setup.py文件，添加以下内容：

from setuptools import setup
from glob import glob

package_name = 'py_topic_demo'

setup(
    name=package_name,
    version='0.0.0',
    packages=[package_name],
    data_files=[
        ('share/ament_index/resource_index/packages',
            ['resource/' + package_name]),
        ('share/' + package_name, ['package.xml']),
        (os.path.join('share', package_name,'msg'), glob('msg/*.msg')),
    ],
    install_requires=['setuptools'],
    zip_safe=True,
    maintainer='your_name',
    maintainer_email='your_email@example.com',
    description='ROS2 topic demo',
    license='Apache License 2.0',
    tests_require=['pytest'],
    entry_points={
        'console_scripts': [
            'publisher = py_topic_demo.publisher:main',
          'subscriber = py_topic_demo.subscriber:main',
        ],
    },
    package_dir={'': 'python'},
    cmake_dependencies=[],
    build_requires=['ament_cmake', ],
    depends=['rclpy','std_msgs','message_generation'],
    classifiers=[
        'Intended Audience :: Developers',
        'License :: OSI Approved :: Apache Software License',
        'Programming Language :: Python',
        'Topic :: Software Development',
    ],
)

同时，编辑setup.py文件，添加以下内容：

from setuptools import setup
from glob import glob

package_name = 'py_topic_demo'

setup(
    name=package_name,
    version='0.0.0',
    packages=[package_name],
    data_files=[
        ('share/ament_index/resource_index/packages',
            ['resource/' + package_name]),
        ('share/' + package_name, ['package.xml']),
        (os.path.join('share', package_name,'msg'), glob('msg/*.msg')),
    ],
    install_requires=['setuptools'],
    zip_safe=True,
    maintainer='your_name',
    maintainer_email='your_email@example.com',
    description='ROS2 topic demo',
    license='Apache License 2.0',
    tests_require=['pytest'],
    entry_points={
        'console_scripts': [
            'publisher = py_topic_demo.publisher:main',
          'subscriber = py_topic_demo.subscriber:main',
        ],
    },
    package_dir={'': 'python'},
    cmake_dependencies=[],
    build_requires=['ament_cmake', ],
    depends=['rclpy','std_msgs','message_generation'],
    classifiers=[
        'Intended Audience :: Developers',
        'License :: OSI Approved :: Apache Software License',
        'Programming Language :: Python',
        'Topic :: Software Development',
    ],
)

4.3 编写发布方代码（publisher_custom.py）

在src目录下创建publisher_custom.py文件，内容如下：

import rclpy
from rclpy.node import Node
from py_topic_demo.msg import Student

class CustomPublisher(Node):
    def __init__(self):
        super().__init__('custom_publisher')
        self.publisher_ = self.create_publisher(Student,'student_topic', 10)
        timer_period = 1  # 发布频率为1Hz
        self.timer = self.create_timer(timer_period, self.timer_callback)
        self.i = 0

    def timer_callback(self):
        msg = Student()
        msg.name = '张三'
        msg.age = 20 + self.i
        msg.height = 1.65 + self.i * 0.01
        self.publisher_.publish(msg)
        self.get_logger().info('发布的学生信息: name=%s, age=%d, height=%.2f' % (msg.name, msg.age, msg.height))
        self.i += 1

def main(args=None):
    rclpy.init(args=args)
    custom_publisher = CustomPublisher()
    rclpy.spin(custom_publisher)
    custom_publisher.destroy_node()
    rclpy.shutdown()

4.4 编写订阅方代码（subscriber_custom.py）

在src目录下创建subscriber_custom.py文件，内容如下：

import rclpy
from rclpy.node import Node
from py_topic_demo.msg import Student

class CustomSubscriber(Node):
    def __init__(self):
        super().__init__('custom_subscriber')
        self.subscription = self.create_subscription(
            Student,
          'student_topic',
            self.listener_callback,
            10)
        self.subscription  # 防止未使用变量警告

    def listener_callback(self, msg):
        self.get_logger().info('订阅的学生消息: name=%s, age=%d, height=%.2f' % (msg.name, msg.age, msg.height))

def main(args=None):
    rclpy.init(args=args)
    custom_subscriber = CustomSubscriber()
    rclpy.spin(custom_subscriber)
    custom_subscriber.destroy_node()
    rclpy.shutdown()

4.5 运行节点

打开两个终端，分别执行以下命令：

终端1（运行发布方）：

source ~/ros2_ws/install/setup.bash
ros2 run py_topic_demo publisher_custom.py

终端2（运行订阅方）：

source ~/ros2_ws/install/setup.bash
ros2 run py_topic_demo subscriber_custom.py

此时，发布方将以1Hz的频率发布自定义的学生信息消息，订阅方将接收到这些消息并打印到终端。

五、总结

通过以上两个案例，我们详细介绍了如何使用Python代码在ROS2中实现简单文本话题通信和自定义接口话题通信。话题通信是ROS2中非常基础和重要的功能，掌握它对于开发复杂的机器人应用具有重要意义。希望本文能对ROS2开发者有所帮助。

DAMO开发者矩阵

DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐

【卡尔曼滤波教程】卡尔曼滤波学习笔记

DAMO开发者矩阵

评审会上说“有风险”，一定要补上这 4 句话

DAMO开发者矩阵

LangChain SummarizationMiddleware：解决上下文腐败的主要手段

DAMO开发者矩阵

所有评论(0)

查看更多评论

Zorion2

@Zorion2

已为社区贡献1条内容

ROS2话题通信Python实现详解

Zorion2

一、引言

二、准备工作

2.1 安装ROS2

2.2 创建工作空间和功能包

三、简单文本话题通信

3.1 编写发布方代码（publisher.py）

3.2 编写订阅方代码（subscriber.py）

3.3 运行节点

四、自定义接口话题通信

4.1 定义自定义接口消息

4.2 配置功能包

4.3 编写发布方代码（publisher_custom.py）

4.4 编写订阅方代码（subscriber_custom.py）

4.5 运行节点

五、总结

所有评论(0)

温馨提示：您尚未绑定手机号

Zorion2