用ROS话题(Topic)和自定义消息，手把手教你搭建一个简易机器人‘聊天室’

用ROS话题搭建多机器人聊天室：从消息设计到私聊过滤实战

想象一下，一群机器人在虚拟空间里自由交谈——有的在公共频道高谈阔论，有的则窃窃私语进行私密对话。这听起来像是科幻场景，但用ROS的话题通信机制配合自定义消息，我们完全可以构建这样一个有趣的"机器人聊天室"系统。不同于传统枯燥的技术演示，这个项目将带你用可视化方式理解ROS的核心通信模型。

1. 聊天室架构设计与消息定制

任何聊天系统的核心都是消息格式的定义。在我们的机器人聊天室中，每条消息需要包含三个关键元素：

# ChatMessage.msg 文件内容 string sender_id # 发送者唯一标识 string content # 消息文本内容 time timestamp # 消息发送时间戳

创建自定义消息的完整流程如下：

1. 在工作包的msg目录下创建ChatMessage.msg文件
2. 修改package.xml添加消息生成依赖：

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

3. 更新CMakeLists.txt配置：

   find_package(catkin REQUIRED COMPONENTS rospy message_generation ) add_message_files(FILES ChatMessage.msg) generate_messages(DEPENDENCIES std_msgs)

编译成功后，可以通过rosmsg show ChatMessage验证消息定义。这种结构化设计使得消息不仅包含内容，还带有完整的元数据，为后续的过滤和路由打下基础。

提示：消息字段命名应保持一致性，建议使用下划线命名法（如sender_id而非senderId）

2. 聊天室主播：消息发布节点实现

我们的"主播"节点需要完成以下功能：

• 定期生成模拟聊天内容
• 支持广播和定向发送两种模式
• 保持稳定的消息发布频率

#!/usr/bin/env python import rospy from chat_room.msg import ChatMessage from random import choice class ChatPublisher: def __init__(self): self.pub = rospy.Publisher('chat_room', ChatMessage, queue_size=10) self.robots = ['R2D2', 'C3PO', 'BB8', 'Wall-E'] self.topics = ['天气', '任务', '笑话', '状态'] def generate_message(self, target=None): msg = ChatMessage() msg.sender_id = choice(self.robots) msg.content = f"关于{choice(self.topics)}：{''.join(['啊']*rospy.get_time()%5)}" msg.timestamp = rospy.get_time() return msg def run(self, rate=1): rate = rospy.Rate(rate) while not rospy.is_shutdown(): msg = self.generate_message() self.pub.publish(msg) rate.sleep() if __name__ == '__main__': rospy.init_node('chat_publisher') ChatPublisher().run(rate=2)

关键参数说明：

使用rostopic echo /chat_room可以实时查看发布的聊天内容。为了更直观观察，可以打开rqt_graph查看节点连接关系：

rqt_graph &

3. 消息订阅与过滤：实现私聊功能

基础订阅者可以接收所有消息，但真正的聊天室需要更精细的控制。我们通过消息过滤实现"私聊"效果：

#!/usr/bin/env python import rospy from chat_room.msg import ChatMessage class ChatSubscriber: def __init__(self, robot_name): self.robot_name = robot_name self.sub = rospy.Subscriber('chat_room', ChatMessage, self.callback) def callback(self, msg): # 基础过滤：只接收特定发送者的消息 if msg.sender_id == self.robot_name: rospy.loginfo(f"[私聊] {msg.sender_id}: {msg.content}") else: rospy.loginfo(f"[广播] {msg.sender_id}: {msg.content}") if __name__ == '__main__': rospy.init_node('chat_subscriber') ChatSubscriber(rospy.get_param('~robot_name', 'R2D2')) rospy.spin()

高级过滤技术对比：

启动多个订阅节点模拟不同机器人：

ROS_NAMESPACE=bb8 rosrun chat_room subscriber.py _robot_name:=BB8 ROS_NAMESPACE=r2d2 rosrun chat_room subscriber.py _robot_name:=R2D2

4. 系统监控与调试技巧

完善的聊天系统需要监控工具保障运行。ROS提供了丰富的命令行工具：

1. 查看活跃话题：

   rostopic list

2. 监控消息流量：

   rostopic hz /chat_room

3. 检查消息结构：

   rosmsg show ChatMessage

4. 可视化工具组合：

   rqt_console & # 查看日志 rqt_graph & # 拓扑关系 rqt_plot & # 数据趋势

常见问题排查指南：

1. 消息未接收：

   • 检查rostopic echo /chat_room是否有输出
   • 确认订阅者节点已正确启动
   • 验证消息类型是否匹配

2. 高延迟：

   rostopic delay /chat_room

   • 减小发布频率
   • 增加queue_size参数

3. 消息丢失：

   • 考虑使用latch模式保留最后消息
   • 检查网络带宽rostopic bw /chat_room

5. 高级功能扩展

基础聊天室搭建完成后，可以考虑以下增强功能：

1. 消息持久化：

   import sqlite3 conn = sqlite3.connect('chat.db') c = conn.cursor() c.execute('''CREATE TABLE IF NOT EXISTS messages (sender text, content text, timestamp real)''')

2. 加密通信：

   from cryptography.fernet import Fernet key = Fernet.generate_key() cipher_suite = Fernet(key) encrypted_msg = cipher_suite.encrypt(msg.content.encode())

3. QoS策略配置：

   from rospy import Publisher pub = Publisher('chat_room', ChatMessage, queue_size=5, subscriber_listener=MyListener())

性能优化参数参考：

6. 实战案例：多机器人协作场景

将聊天室概念扩展到实际机器人协作，例如：

1. 任务分配系统：

   def task_callback(msg): if "任务" in msg.content: if available_resources(): accept_task(msg.sender_id)

2. 异常报警网络：

   class EmergencyMonitor: def __init__(self): self.sub = rospy.Subscriber('alerts', AlertMsg, self.handle_alert) def handle_alert(self, msg): if msg.level > 3: notify_all_robots(msg)

3. 数据共享平台：

   class DataHub: def __init__(self): self.data_store = {} self.sub = rospy.Subscriber('data_share', DataMsg, self.store) def store(self, msg): self.data_store[msg.key] = msg.value

在Gazebo仿真环境中测试时，可以通过以下命令观察通信效果：

roslaunch gazebo_ros empty_world.launch rosrun chat_room publisher.py _rate:=5 rostopic echo /chat_room -n 10

7. 工程化建议��最佳实践

将原型转化为可维护的系统需要：

1. 代码组织规范：

   chat_room/ ├── launch/ # 启动文件 ├── msg/ # 自定义消息 ├── scripts/ # Python节点 ├── src/ # C++节点 └── test/ # 测试用例

2. 日志记录策略：

   import logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')

3. 异常处理模式：

   try: msg = receive_message() except rospy.ROSInterruptException: cleanup() except MessageParseError as e: log_error(e)

性能基准测试结果示例：