本文用最通俗的方式讲清 ROS2 最核心的三个概念:节点、话题、服务。读完你就能动手写第一个 ROS2 项目。适合零基础入门,代码全在文末,直接复制就能跑。
📌 为什么要学 ROS2?
ROS2(Robot Operating System 2)是目前机器人开发领域的事实标准。不管你是做无人机、机械臂、智能小车,还是工业机器人,ROS2 都是绕不开的技术栈。
但很多新手一开始就被各种概念劝退了:节点、话题、服务、动作、参数、DDS…… 别急,核心概念其实只有三个:节点、话题、服务。把这三个搞明白,ROS2 就算入门了。
先上一张架构图,帮你建立全局认知:
下面我们一个个来拆。
🔹 一、节点(Node):机器人的"器官"
什么是节点?
节点就是一个独立的功能模块,负责做一件具体的事。
比如:
- 一个节点负责读取摄像头图像
- 一个节点负责处理激光雷达数据
- 一个节点负责路径规划
- 一个节点负责控制电机转动
每个节点各司其职,通过 ROS2 的通信机制互相配合,就像人体的各个器官协同工作一样。
💡通俗理解:节点就像一个公司里的员工,每个人只干自己的活,但大家通过"聊天"(通信)来协作完成一个大项目。
为什么要用节点?
- 模块化:每个功能独立,出问题好找原因
- 可复用:同一个节点可以在不同项目里用
- 分布式:节点可以跑在不同电脑上,甚至不同设备上
动手写第一个节点
我们用 Python 写一个最简单的节点,它什么都不做,就是活着打印一条日志:
importrclpyfromrclpy.nodeimportNodeclassMyFirstNode(Node):"""我的第一个 ROS2 节点"""def__init__(self):# 给节点起个名字:my_first_nodesuper().__init__('my_first_node')# 调用父类的初始化self.get_logger().info('我的第一个 ROS2 节点启动了!')defmain(args=None):# 初始化 ROS2rclpy.init(args=args)# 创建节点实例node=MyFirstNode()# 让节点保持运行(进入自旋循环)相当于while(1)不断处理回调rclpy.spin(node)# 关闭时清理rclpy.shutdown()if__name__=='__main__':main()运行效果:
[INFO] [my_first_node]: 我的第一个 ROS2 节点启动了!节点跑起来了,但一个节点没啥意思。节点的价值在于和其他节点通信。这就引出了第二个核心概念:话题。
🔹 二、话题(Topic):节点间的"广播电台"
什么是话题?
话题是 ROS2 里最常用的通信方式,采用发布-订阅模式(Publish-Subscribe)。
- 发布者(Publisher):往话题里发消息的节点
- 订阅者(Subscriber):从话题里收消息的节点
- 话题(Topic):消息的"频道",用名字标识,比如
/camera/image
一个话题可以有多个发布者,也可以有多个订阅者。发布者不需要知道谁在订阅,订阅者也不需要知道谁在发布——完全解耦。
💡通俗理解:话题就像广播电台。电台(发布者)只管往外播节目,收音机(订阅者)调到对应频率就能听。电台不知道有多少人在听,听众也不需要认识电台主播。
话题的特点
| 特点 | 说明 |
|---|---|
| 单向通信 | 发布者只管发,订阅者只管收 |
| 一对多 | 一个话题可以被 N 个订阅者同时接收 |
| 持续流 | 适合高频、连续的数据(传感器、控制指令) |
| 异步 | 发了就走,不等待回应 |
实战:写一个"温度传感器 + 显示器"
我们做两个节点:
temperature_publisher:模拟温度传感器,每秒发布一次温度数据temperature_subscriber:订阅温度数据,显示在屏幕上
① 发布者节点
importrclpyfromrclpy.nodeimportNodefromstd_msgs.msgimportFloat32importrandomclassTemperaturePublisher(Node):"""温度传感器发布者节点"""def__init__(self):super().__init__('temperature_publisher')# 创建发布者:消息类型 Float32,话题名 /temperature,队列大小 10self.publisher_=self.create_publisher(Float32,'/temperature',10)# 创建定时器,每 1 秒发布一次self.timer=self.create_timer(1.0,self.publish_temperature)self.get_logger().info('🌡️ 温度传感器节点已启动')defpublish_temperature(self):"""发布温度数据"""# 模拟温度:25°C 左右浮动temp=25.0+random.uniform(-2.0,2.0)msg=Float32()msg.data=round(temp,2)self.publisher_.publish(msg)self.get_logger().info(f'发布温度:{msg.data}°C')defmain(args=None):rclpy.init(args=args)node=TemperaturePublisher()rclpy.spin(node)rclpy.shutdown()if__name__=='__main__':main()② 订阅者节点
importrclpyfromrclpy.nodeimportNodefromstd_msgs.msgimportFloat32classTemperatureSubscriber(Node):"""温度显示订阅者节点"""def__init__(self):super().__init__('temperature_subscriber')# 创建订阅者:话题 /temperature,回调函数 temperature_callbackself.subscription=self.create_subscription(Float32,'/temperature',self.temperature_callback,10# 队列大小)self.get_logger().info('📺 温度显示节点已启动,等待数据...')deftemperature_callback(self,msg):"""收到温度数据后的回调函数"""temp=msg.data# 根据温度给不同提示iftemp>28:status='🔥 有点热'eliftemp<22:status='❄️ 有点凉'else:status='✅ 舒适'self.get_logger().info(f'收到温度:{temp}°C{status}')defmain(args=None):rclpy.init(args=args)node=TemperatureSubscriber()rclpy.spin(node)rclpy.shutdown()if__name__=='__main__':main()运行效果:
# 终端1:运行发布者 $ ros2 run my_package temperature_publisher [INFO] [temperature_publisher]: 🌡️ 温度传感器节点已启动 [INFO] [temperature_publisher]: 发布温度: 24.35 °C [INFO] [temperature_publisher]: 发布温度: 26.12 °C # 终端2:运行订阅者 $ ros2 run my_package temperature_subscriber [INFO] [temperature_subscriber]: 📺 温度显示节点已启动,等待数据... [INFO] [temperature_subscriber]: 收到温度: 24.35 °C ✅ 舒适 [INFO] [temperature_subscriber]: 收到温度: 26.12 °C ✅ 舒适你看,两个节点完全独立,只是通过/temperature这个话题就完成了通信,非常优雅。
🔹 三、服务(Service):节点间的"问答对话"
什么是服务?
服务是 ROS2 的第二种通信方式,采用请求-响应模式(Request-Response)。
- 服务端(Server):提供服务的节点,收到请求后处理并返回响应
- 客户端(Client):发起请求的节点,等待服务端回应
- 服务(Service):用名字标识,比如
/add_two_ints
和话题不同,服务是双向的:客户端发一个请求,服务端必须返回一个响应。
💡通俗理解:服务就像打电话问问题。你(客户端)打给客服(服务端),问一个问题(请求),客服查一下然后告诉你答案(响应)。是一对一的、有问有答的交互。
服务 vs 话题,怎么选?
| 对比项 | 话题(Topic) | 服务(Service) |
|---|---|---|
| 通信模式 | 发布-订阅(单向) | 请求-响应(双向) |
| 数据流向 | 单向 | 一来一回 |
| 适用场景 | 传感器数据、状态广播 | 触发动作、查询状态 |
| 实时性 | 高(持续流) | 中(一问一答) |
| 可靠性 | 可能丢数据 | 保证有回应 |
| 类比 | 广播电台 | 打电话 / REST API |
简单判断:
- 你需要持续不断发数据→ 用话题
- 你需要调用一个功能并拿到结果→ 用服务
实战:写一个"加法计算器"服务
我们做两个节点:
add_server:提供加法服务,接收两个数,返回它们的和add_client:调用加法服务,传入两个数,打印结果
① 服务端节点
importrclpyfromrclpy.nodeimportNodefromexample_interfaces.srvimportAddTwoIntsclassAddTwoIntsServer(Node):"""加法服务端节点"""def__init__(self):super().__init__('add_two_ints_server')# 创建服务:服务类型 AddTwoInts,服务名 /add_two_intsself.srv=self.create_service(AddTwoInts,'/add_two_ints',self.add_two_ints_callback)self.get_logger().info('🧮 加法服务已就绪,等待请求...')defadd_two_ints_callback(self,request,response):"""处理服务请求的回调函数"""# 从请求中取出两个数a=request.a b=request.b# 计算结果response.sum=a+b self.get_logger().info(f'收到请求:{a}+{b}={response.sum}')# 返回响应returnresponsedefmain(args=None):rclpy.init(args=args)node=AddTwoIntsServer()rclpy.spin(node)rclpy.shutdown()if__name__=='__main__':main()② 客户端节点
importrclpyfromrclpy.nodeimportNodefromexample_interfaces.srvimportAddTwoIntsimportsysclassAddTwoIntsClient(Node):"""加法客户端节点"""def__init__(self):super().__init__('add_two_ints_client')# 创建客户端:服务类型 AddTwoInts,服务名 /add_two_intsself.cli=self.create_client(AddTwoInts,'/add_two_ints')# 等待服务上线(最多等 5 秒)whilenotself.cli.wait_for_service(timeout_sec=1.0):self.get_logger().info('⏳ 等待服务上线...')self.get_logger().info('✅ 服务已连接')defsend_request(self,a,b):"""发送加法请求"""req=AddTwoInts.Request()req.a=a req.b=b# 异步发送请求self.future=self.cli.call_async(req)# 等待响应rclpy.spin_until_future_complete(self,self.future)returnself.future.result()defmain(args=None):rclpy.init(args=args)client=AddTwoIntsClient()# 从命令行参数读取两个数,默认 3 和 5a=int(sys.argv[1])iflen(sys.argv)>1else3b=int(sys.argv[2])iflen(sys.argv)>2else5response=client.send_request(a,b)client.get_logger().info(f'📤 请求:{a}+{b}')client.get_logger().info(f'📥 结果:{response.sum}')rclpy.shutdown()if__name__=='__main__':main()运行效果:
# 终端1:启动服务端 $ ros2 run my_package add_server [INFO] [add_two_ints_server]: 🧮 加法服务已就绪,等待请求... # 终端2:运行客户端 $ ros2 run my_package add_client 10 20 [INFO] [add_two_ints_client]: ✅ 服务已连接 [INFO] [add_two_ints_client]: 📤 请求: 10 + 20 [INFO] [add_two_ints_client]: 📥 结果: 30 # 服务端会打印 [INFO] [add_two_ints_server]: 收到请求: 10 + 20 = 30🔹 四、三大概念一张表总结
| 概念 | 作用 | 模式 | 类比 | 适用场景 |
|---|---|---|---|---|
| 节点 Node | 功能模块 | - | 员工 / 器官 | 每个独立功能 |
| 话题 Topic | 数据流通信 | 发布-订阅 | 广播电台 | 传感器、状态、高频数据 |
| 服务 Service | 请求式通信 | 请求-响应 | 打电话 / API | 触发动作、查询、配置 |
🚀 下一步学什么?
掌握了节点、话题、服务,ROS2 的骨架就搭起来了。接下来可以继续学:
- 动作(Action):适合耗时任务(比如机械臂运动、导航),是服务的增强版,支持中途反馈和取消
- 参数(Parameter):节点的配置项,可以运行时动态修改
- TF 坐标变换:机器人各部件的位置关系
- 导航 Navigation2:自主移动机器人的完整导航方案
- MoveIt2:机械臂运动规划
📝 完整代码
本文所有代码都可以直接复制运行,前提是你已经装好 ROS2(推荐 Humble 或 Jazzy 版本)。
如果需要完整的 ROS2 学习路线图、更多实战项目源码,可以关注我,后续持续更新。
点赞 + 收藏 + 关注,学 ROS2 不迷路!有问题欢迎评论区交流~
本文为原创内容,转载请注明出处。