编写简单的发布者和订阅者（Python）

目标： 使用 Python 创建并运行发布者（publisher）和订阅者（subscriber）节点。

教程级别： 初级

预计时间： 20 分钟

背景

在本教程中，你将创建节点，这些节点以字符串消息的形式通过话题相互传递信息。这里使用的示例是一个简单的“发布者/订阅者”系统：一个节点发布数据，另一个节点订阅该话题以接收数据。

本示例中使用的代码可以在这里找到。

前提条件

在之前的教程中，你已经学习了如何创建工作空间和如何创建功能包。

建议你对 Python 有基本的了解，但并非必需。

任务

1 创建功能包

打开一个新终端并 source ROS 2 安装环境，以便使用 ros2 命令。

导航到之前教程中创建的 ros2_ws 目录。

功能包应在 src 目录中创建，而非工作空间根目录。因此，进入 ros2_ws/src，然后运行创建功能包的命令：

ros2 pkg create --build-type ament_python --license Apache-2.0 py_pubsub

终端将返回一条消息，确认功能包 py_pubsub 及其所有必要的文件和文件夹已创建。

2 编写发布者节点

进入 ros2_ws/src/py_pubsub/py_pubsub。该目录是一个 Python 包，与其嵌套的 ROS 2 功能包同名。

输入以下命令下载示例发布者代码：

Linux/macOS:

wget https://raw.githubusercontent.com/ros2/examples/{REPOS_FILE_BRANCH}/rclpy/topics/minimal_publisher/examples_rclpy_minimal_publisher/publisher_member_function.py

Windows:

在 Windows 命令提示符中：

curl -sk https://raw.githubusercontent.com/ros2/examples/{REPOS_FILE_BRANCH}/rclpy/topics/minimal_publisher/examples_rclpy_minimal_publisher/publisher_member_function.py -o publisher_member_function.py

或在 PowerShell 中：

curl https://raw.githubusercontent.com/ros2/examples/{REPOS_FILE_BRANCH}/rclpy/topics/minimal_publisher/examples_rclpy_minimal_publisher/publisher_member_function.py -o publisher_member_function.py

现在在 __init__.py 旁边会有一个名为 publisher_member_function.py 的新文件。

用你喜欢的文本编辑器打开该文件。

import rclpy
from rclpy.executors import ExternalShutdownException
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):
    try:
        with rclpy.init(args=args):
            minimal_publisher = MinimalPublisher()

            rclpy.spin(minimal_publisher)
    except (KeyboardInterrupt, ExternalShutdownException):
        pass


if __name__ == '__main__':
    main()

2.1 代码解析

注释之后的前几行代码导入了 rclpy，以便使用其 Node 类。

import rclpy
from rclpy.executors import ExternalShutdownException
from rclpy.node import Node

下一条语句导入了内置的 std_msgs/msg/String 消息类型，节点使用它来组织在话题上传递的数据结构。

from std_msgs.msg import String

这些 import 语句代表了节点的依赖。注意，依赖必须添加到 package.xml 中，你将在下一节完成这一操作。

接下来，创建 MinimalPublisher 类，它继承自（即子类）Node。

class MinimalPublisher(Node):

下面是类构造函数的定义。super().__init__ 调用 Node 类的构造函数，并传入节点名称，本例中为 minimal_publisher。

create_publisher 声明该节点在名为 topic 的话题上发布 std_msgs/msg/String 类型（从 std_msgs.msg 模块导入）的消息，并且“队列大小”为 10。队列大小是一个必需的服务质量（QoS）设置，用于在订阅者接收速度不够快时限制排队消息的数量。

接下来，通过 create_timer 创建一个每 0.5 秒执行一次的定时回调。self.i 是回调中使用的计数器。

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

timer_callback 创建一条包含计数器值的消息，将其发布，并通过 get_logger() 的 info() 函数将其打印到控制台。

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

最后，定义了 main 函数。

def main(args=None):
    try:
        with rclpy.init(args=args):
            minimal_publisher = MinimalPublisher()

            rclpy.spin(minimal_publisher)
    except (KeyboardInterrupt, ExternalShutdownException):
        pass

首先初始化 rclpy 库，然后创建节点，最后通过 spin() 持续处理该节点的回调。

2.2 添加依赖

返回上一级到 ros2_ws/src/py_pubsub 目录，那里已经为你创建了 setup.py、setup.cfg 和 package.xml 文件。用文本编辑器打开 package.xml。

如之前教程中所述，确保填写 <description>、<maintainer> 和 <license> 标签：

<description>Examples of minimal publisher/subscriber using rclpy</description>
<maintainer email="you@email.com">Your Name</maintainer>
<license>Apache-2.0</license>

在上述行之后，添加与节点 import 语句相对应的以下依赖：

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

这声明了该功能包在执行代码时需要 rclpy 和 std_msgs。

确保保存文件。

2.3 添加入口点

打开 setup.py 文件。同样，将 maintainer、maintainer_email、description 和 license 字段与 package.xml 保持一致：

maintainer='YourName',
maintainer_email='you@email.com',
description='Examples of minimal publisher/subscriber using rclpy',
license='Apache-2.0',

在 entry_points 字段的 console_scripts 列表中添加以下行：

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

别忘了保存。

2.4 检查 setup.cfg

setup.cfg 文件的内容应该会自动正确填充，如下所示：

[develop]
script_dir=$base/lib/py_pubsub
[install]
install_scripts=$base/lib/py_pubsub

这只是告诉 setuptools 将可执行文件放在 lib 中，因为 ros2 run 会在那里查找它们。

你现在可以构建功能包、source 本地 setup 文件并运行它，但让我们先创建订阅者节点，这样就能看到整个系统运行的效果。

3 编写订阅者节点

返回 ros2_ws/src/py_pubsub/py_pubsub 创建下一个节点。在终端中输入以下命令：

Linux/macOS:

wget https://raw.githubusercontent.com/ros2/examples/{REPOS_FILE_BRANCH}/rclpy/topics/minimal_subscriber/examples_rclpy_minimal_subscriber/subscriber_member_function.py

Windows:

在 Windows 命令提示符中：

curl -sk https://raw.githubusercontent.com/ros2/examples/{REPOS_FILE_BRANCH}/rclpy/topics/minimal_subscriber/examples_rclpy_minimal_subscriber/subscriber_member_function.py -o subscriber_member_function.py

或在 PowerShell 中：

curl https://raw.githubusercontent.com/ros2/examples/{REPOS_FILE_BRANCH}/rclpy/topics/minimal_subscriber/examples_rclpy_minimal_subscriber/subscriber_member_function.py -o subscriber_member_function.py

现在该目录应该包含以下文件：

__init__.py  publisher_member_function.py  subscriber_member_function.py

3.1 代码解析

用文本编辑器打开 subscriber_member_function.py。

import rclpy
from rclpy.executors import ExternalShutdownException
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):
    try:
        with rclpy.init(args=args):
            minimal_subscriber = MinimalSubscriber()

            rclpy.spin(minimal_subscriber)
    except (KeyboardInterrupt, ExternalShutdownException):
        pass


if __name__ == '__main__':
    main()

订阅者节点的代码与发布者的几乎完全相同。构造函数使用 create_subscription 创建一个订阅者，参数与发布者相同。注意，发布者和订阅者使用的话题名称和消息类型必须匹配，双方才能进行通信。

self.subscription = self.create_subscription(
    String,
    'topic',
    self.listener_callback,
    10)

订阅者的构造函数和回调不包含任何 timer 定义，因为不需要。一旦收到消息，回调就会被调用。

回调函数简单地将一条 info 消息连同接收到的数据打印到控制台。注意，发布者定义了 msg.data = 'Hello World: %d' % self.i

def listener_callback(self, msg):
    self.get_logger().info('I heard: "%s"' % msg.data)

main 的定义几乎完全相同，只是将发布者的创建和 spin 替换为订阅者。

minimal_subscriber = MinimalSubscriber()

rclpy.spin(minimal_subscriber)

由于该节点与发布者具有相同的依赖，因此不需要在 package.xml 中添加任何新内容。setup.cfg 文件也可以保持不变。

3.2 添加入口点

重新打开 setup.py，在发布者的入口点下方为订阅者节点添加入口点。entry_points 字段现在应该如下所示：

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

保存文件后，你的发布/订阅系统就准备好了。

4 构建与运行

你的 ROS 2 系统中可能已经安装了 rclpy 和 std_msgs 功能包。在构建之前，建议在工作空间（ros2_ws）的根目录下运行 rosdep 来检查缺失的依赖：

Linux:

rosdep install -i --from-path src --rosdistro {DISTRO} -y

macOS:

rosdep 仅在 Linux 上运行，可直接跳到下一步。

Windows:

rosdep 仅在 Linux 上运行，可直接跳到下一步。

仍在工作空间根目录 ros2_ws 下构建新功能包：

Linux/macOS:

colcon build --packages-select py_pubsub

Windows:

colcon build --merge-install --packages-select py_pubsub

打开一个新终端，导航到 ros2_ws，并 source 环境配置文件：

Linux:

source install/setup.bash

macOS:

. install/setup.bash

Windows:

call install/setup.bat

现在运行 talker 节点。终端应该每 0.5 秒发布一条 info 消息，如下所示：

$ ros2 run py_pubsub talker
[info] [minimal_publisher]: publishing: "hello world: 0"
[info] [minimal_publisher]: publishing: "hello world: 1"
[info] [minimal_publisher]: publishing: "hello world: 2"
[info] [minimal_publisher]: publishing: "hello world: 3"
[info] [minimal_publisher]: publishing: "hello world: 4"
...

打开另一个终端，再次从 ros2_ws 中 source 环境配置文件，然后启动 listener 节点。listener 将开始向控制台打印消息，从发布者当时的消息计数开始，如下所示：

$ ros2 run py_pubsub listener
[INFO] [minimal_subscriber]: I heard: "Hello World: 10"
[INFO] [minimal_subscriber]: I heard: "Hello World: 11"
[INFO] [minimal_subscriber]: I heard: "Hello World: 12"
[INFO] [minimal_subscriber]: I heard: "Hello World: 13"
[INFO] [minimal_subscriber]: I heard: "Hello World: 14"

在每个终端中输入 Ctrl+C 停止节点的运行。

总结

你创建了两个节点，通过话题发布和订阅数据。在运行它们之前，你将它们的依赖和入口点添加到了功能包配置文件中。

下一步

接下来，你将使用服务/客户端模型创建另一个简单的 ROS 2 功能包。同样，你可以选择用 C++ 或 Python 来编写。

编写简单的发布者和订阅者（Python）

背景

前提条件

任务

1 创建功能包

2 编写发布者节点

2.1 代码解析

2.2 添加依赖

2.3 添加入口点

2.4 检查 setup.cfg

3 编写订阅者节点

3.1 代码解析

3.2 添加入口点

4 构建与运行

总结

下一步

相关内容