ROS2系列 (9) : Python话题通信节点——订阅者示例

在上一篇教程中，我们实现了“定时发布小说文本”的话题发布者。本文将通过订阅者节点实现“接收小说并逐行朗读”的功能，完成“发布-订阅”的完整闭环。我们会将该节点集成到同一个工作空间，实现从“文本发布”到“语音朗读”的端到端流程。

一、需求与技术拆解

鱼老板需求：订阅小说话题，逐行合成语音并朗读。

需解决的核心问题：

• 如何订阅话题？—— 使用rclpy的create_subscription创建订阅者。
• 如何朗读文本？—— 借助espeak-ng库实现中文语音合成。
• 如何处理“发布快、朗读慢”的异步问题？—— 用Queue队列做缓冲，结合多线程分离“接收”与“朗读”逻辑。

二、环境准备与功能包集成

2.1 安装语音合成依赖

安装espeak-ng的Python库和系统依赖：

# 安装Python库
pip3 install espeakng
# 安装系统依赖（语音合成引擎）
sudo apt install espeak-ng

2.2 集成到现有工作空间

将订阅者节点功能包添加到上一篇的工作空间ros2_ws/src中：

cd ~/ros2_ws/src
# 创建订阅者功能包
ros2 pkg create demo_python_topic_sub --build-type ament_python \
  --dependencies rclpy example_interfaces espeakng \
  --license Apache-2.0

三、配置功能包依赖

3.1 编辑package.xml

打开demo_python_topic_sub/package.xml，添加依赖：

<?xml version="1.0"?>
<?xml-model href="http://download.ros.org/schema/package_format3.xsd" schematypens="http://www.w3.org/2001/XMLSchema"?>
<package format="3">
  <name>demo_python_topic_sub</name>
  <version>0.0.0</version>
  <description>订阅小说话题并朗读</description>
  <maintainer email="your@email.com">your_name</maintainer>
  <license>Apache-2.0</license>

  <!-- 核心依赖 -->
  <depend>rclpy</depend>
  <depend>example_interfaces</depend>
  <depend>espeakng</depend>  <!-- 语音合成依赖 -->

  <buildtool_depend>ament_python</buildtool_depend>
  <test_depend>ament_copyright</test_depend>
  <test_depend>ament_flake8</test_depend>
  <test_depend>ament_pep257</test_depend>
  <test_depend>pytest</test_depend>

  <export>
    <build_type>ament_python</build_type>
  </export>
</package>

3.2 编辑setup.py（注册节点入口）

打开demo_python_topic_sub/setup.py，添加入口：

from setuptools import setup

package_name = 'demo_python_topic_sub'

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']),
    ],
    install_requires=['setuptools', 'espeakng'],
    zip_safe=True,
    maintainer='your_name',
    maintainer_email='your@email.com',
    description='订阅小说话题并朗读',
    license='Apache-2.0',
    tests_require=['pytest'],
    entry_points={
        'console_scripts': [
            'novel_read = demo_python_topic_sub.novel_sub:main',
        ],
    },
)

四、编写订阅者节点代码

在demo_python_topic_sub/demo_python_topic_sub/目录下创建novel_sub.py：

import rclpy
from rclpy.node import Node
from example_interfaces.msg import String
import threading
from queue import Queue
import time
import espeakng


class NovelSubNode(Node):
    def __init__(self, node_name):
        super().__init__(node_name)
        # 1. 初始化队列（缓冲小说行，解决“发布快、朗读慢”的异步问题）
        self.novels_queue_ = Queue()
        # 2. 创建话题订阅者：订阅`novel`话题，消息类型String，队列长度10
        self.novel_subscriber_ = self.create_subscription(
            String, 'novel', self.novel_callback, 10)
        self.get_logger().info('话题订阅者已创建，开始监听/novel话题')
        # 3. 启动朗读线程（分离“接收”和“朗读”逻辑）
        self.speech_thread_ = threading.Thread(target=self.speak_thread)
        self.speech_thread_.daemon = True  # 线程随节点退出而终止
        self.speech_thread_.start()

    def novel_callback(self, msg):
        """话题回调：接收小说行并放入队列"""
        self.novels_queue_.put(msg.data)
        self.get_logger().info(f'已接收小说行：{msg.data}')

    def speak_thread(self):
        """朗读线程：从队列取文本并合成语音"""
        speaker = espeakng.Speaker()
        speaker.voice = 'zh'  # 设置为中文语音
        self.get_logger().info('语音合成引擎已初始化，开始等待朗读任务')
        while rclpy.ok():  # 节点运行时持续执行
            if not self.novels_queue_.empty():
                text = self.novels_queue_.get()
                self.get_logger().info(f'正在朗读：{text}')
                speaker.say(text)
                speaker.wait()  # 等待朗读完成
            else:
                time.sleep(1)  # 队列为空时休眠，减少CPU占用


def main(args=None):
    rclpy.init(args=args)  # 初始化ROS2
    node = NovelSubNode('novel_read_node')  # 创建节点
    rclpy.spin(node)  # 保持节点运行
    rclpy.shutdown()  # 关闭ROS2

五、代码关键模块解析

5.1 话题订阅者（Subscriber）

create_subscription(msg_type, topic_name, callback, qos_profile)用于创建订阅者：

• msg_type：消息类型（本案例中为example_interfaces.msg.String）。
• topic_name：订阅的话题名称（需与发布者的novel话题一致）。
• callback：接收到消息时执行的函数（本案例中为novel_callback）。
• qos_profile：服务质量配置（本案例中为10）。

5.2 多线程与队列的协同

• 队列（Queue）：作为“接收”和“朗读”的缓冲，避免因朗读速度慢导致消息丢失。
• 多线程：speech_thread独立于ROS2主循环运行，确保朗读逻辑不阻塞话题接收。

5.3 语音合成（espeak-ng）

espeakng.Speaker()提供跨平台的语音合成能力：

• speaker.voice = 'zh'：设置为中文语音（支持多语言切换，如en为英文）。
• speaker.say(text)：合成并播放文本语音。
• speaker.wait()：等待朗读完成，保证逐行朗读的顺序性。

六、构建与运行完整流程

6.1 构建整个工作空间

回到工作空间根目录，构建所有功能包：

cd ~/ros2_ws
colcon build

6.2 启动完整流程（三个终端）

终端1：启动本地小说服务器（若已关闭需重新启动）

cd ~/ros2_ws/novel_server
python3 -m http.server

终端2：启动小说发布者节点

cd ~/ros2_ws
source install/setup.bash
ros2 run demo_python_topic novel_pub

终端3：启动小说订阅与朗读节点

cd ~/ros2_ws
source install/setup.bash
ros2 run demo_python_topic_sub novel_read

6.3 效果验证

• 发布者终端会每隔5秒输出“发布了一行小说”；
• 订阅者终端会输出“已接收小说行”和“正在朗读”，同时扬声器会逐行朗读小说内容。

七、常见问题解决

1. 语音无输出：检查espeak-ng是否安装成功，可在终端执行espeak-ng "测试语音"验证。
2. 订阅不到话题：确保发布者和订阅者的话题名称（novel）一致，且已source相同的环境变量。
3. 队列阻塞/溢出：若朗读速度远慢于发布速度，可增大队列长度或优化朗读逻辑（如批量朗读）。

总结

本文通过“小说朗读”案例，完整实现了ROS2话题通信的订阅者逻辑，并与上一篇的发布者形成闭环。核心技术点包括：

• 话题订阅者的创建与回调逻辑；
• 多线程+队列解决异步数据处理问题；
• espeak-ng实现中文语音合成。

掌握这些内容后，你可以轻松实现各类“数据接收+业务处理”的ROS2节点，如传感器数据解析、指令响应等场景。