2、ROS2核心概念深度解析
各位同学，欢迎来到第二章。上一章我们搭好了环境，这一章咱们来啃硬骨头——ROS2的四大核心概念。说实话，很多初学者就是卡在这一关，觉得节点、话题、服务、动作这些概念太抽象。别急，我用实际项目中的经验帮你把它们拆开揉碎。
2.1 节点（Node）：ROS2的"细胞"
节点是什么？说白了，节点就是ROS2系统里最小的独立执行单元。每个节点负责一个具体的功能，比如读取激光雷达数据、控制电机、处理图像。我个人习惯把节点想象成一个乐高积木——每个积木干一件事，拼在一起就成了机器人系统。
2.1.1 创建节点
在ROS2里创建节点，Python和C++都支持。我主要用Python做原型验证，C++做性能敏感的部分。先看Python版本：
import rclpy
from rclpy.node import Node
class MyRobotNode(Node):
def __init__(self):
super().__init__('my_robot_node')
self.get_logger().info('节点已启动！')
def main(args=None):
rclpy.init(args=args)
node = MyRobotNode()
rclpy.spin(node) # 保持节点运行
node.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
我的经验：节点名称在整个ROS2系统中必须唯一。我曾经在一个项目里不小心重复命名了两个节点，结果调试了一下午才发现是命名冲突。建议用
功能_模块_节点的命名方式，比如navigation_path_planner。
2.1.2 节点生命周期
ROS2的节点生命周期比ROS1复杂，但更灵活。它定义了四个主要状态：
| 状态 | 说明 | 触发条件 |
|---|---|---|
| Unconfigured | 未配置，节点刚创建 | 节点实例化 |
| Inactive | 已配置但未激活 | 调用configure() |
| Active | 正常运行中 | 调用activate() |
| Finalized | 已销毁 | 调用cleanup()或shutdown() |
避坑指南：我曾经在真机部署时，因为没处理好节点生命周期，导致机器人启动时传感器还没初始化就开始接收数据，直接报错。记住：一定要在
Active状态之前完成所有初始化工作。
2.2 话题（Topic）：异步通信的"广播站"
话题是ROS2里最常用的通信方式。它就像广播电台——发布者只管发，订阅者只管收，双方不需要知道对方的存在。你想想看，这种解耦设计让系统扩展变得特别容易。
2.2.1 发布者实现
class TalkerNode(Node):
def __init__(self):
super().__init__('talker')
self.publisher = self.create_publisher(String, 'chatter', 10)
self.timer = self.create_timer(0.5, self.timer_callback)
def timer_callback(self):
msg = String()
msg.data = 'Hello, ROS2!'
self.publisher.publish(msg)
self.get_logger().info(f'发布: {msg.data}')2.2.2 订阅者实现
class ListenerNode(Node):
def __init__(self):
super().__init__('listener')
self.subscription = self.create_subscription(
String, 'chatter', self.listener_callback, 10)
def listener_callback(self, msg):
self.get_logger().info(f'收到: {msg.data}')
关键点：话题的QoS（Quality of Service）设置很重要。默认的
10是队列深度，但实际项目中要根据场景调整。比如激光雷达数据需要Best Effort可靠性，而控制指令必须用Reliable。
2.3 服务（Service）：同步通信的"请求-应答"
服务是同步的——客户端发请求，服务器处理完再返回结果。这就像你去柜台办事，必须等工作人员处理完才能走。我一般在需要确认结果的场景用服务，比如"设置目标点"或"查询状态"。
2.3.1 服务端实现
from example_interfaces.srv import AddTwoInts
class AddTwoIntsServer(Node):
def __init__(self):
super().__init__('add_two_ints_server')
self.srv = self.create_service(AddTwoInts, 'add_two_ints', self.add_callback)
def add_callback(self, request, response):
response.sum = request.a + request.b
self.get_logger().info(f'计算: {request.a} + {request.b} = {response.sum}')
return response2.3.2 客户端实现
class AddTwoIntsClient(Node):
def __init__(self):
super().__init__('add_two_ints_client')
self.cli = self.create_client(AddTwoInts, 'add_two_ints')
while not self.cli.wait_for_service(timeout_sec=1.0):
self.get_logger().warn('等待服务...')
def send_request(self, a, b):
req = AddTwoInts.Request()
req.a = a
req.b = b
future = self.cli.call_async(req)
rclpy.spin_until_future_complete(self, future)
return future.result()
我的建议：服务调用一定要加超时处理。我在调试机械臂时，有一次服务端卡死，客户端一直等，整个系统都hang住了。加个
timeout参数，优雅处理异常情况。
2.4 动作（Action）：带反馈的"长期任务"
动作是ROS2里最强大的通信机制。它结合了话题的异步性和服务的反馈能力。比如让机器人走到某个点——你需要知道它走到哪了（反馈），还能随时取消（可抢占）。
2.4.1 动作定义
# Fibonacci.action
int32 order
---
int32[] sequence
---
int32[] partial_sequence2.4.2 动作服务端
class FibonacciActionServer(Node):
def __init__(self):
super().__init__('fibonacci_action_server')
self.action_server = ActionServer(
self, Fibonacci, 'fibonacci', self.execute_callback)
async def execute_callback(self, goal_handle):
feedback_msg = Fibonacci.Feedback()
feedback_msg.partial_sequence = [0, 1]
for i in range(1, goal_handle.request.order):
feedback_msg.partial_sequence.append(
feedback_msg.partial_sequence[i] +
feedback_msg.partial_sequence[i-1])
goal_handle.publish_feedback(feedback_msg)
await asyncio.sleep(0.5) # 模拟耗时
goal_handle.succeed()
result = Fibonacci.Result()
result.sequence = feedback_msg.partial_sequence
return result2.4.3 动作客户端
class FibonacciActionClient(Node):
def __init__(self):
super().__init__('fibonacci_action_client')
self.action_client = ActionClient(self, Fibonacci, 'fibonacci')
def send_goal(self, order):
goal_msg = Fibonacci.Goal()
goal_msg.order = order
self.action_client.wait_for_server()
send_goal_future = self.action_client.send_goal_async(
goal_msg, feedback_callback=self.feedback_callback)
send_goal_future.add_done_callback(self.goal_response_callback)
def feedback_callback(self, feedback_msg):
self.get_logger().info(f'进度: {feedback_msg.feedback.partial_sequence}')
def goal_response_callback(self, future):
goal_handle = future.result()
if not goal_handle.accepted:
self.get_logger().info('目标被拒绝')
return
result_future = goal_handle.get_result_async()
result_future.add_done_callback(self.get_result_callback)
避坑指南：动作的反馈频率要控制好。我曾经在导航任务里每10ms发一次反馈，结果反馈消息把带宽占满了，控制指令反而延迟了。建议反馈频率不超过10Hz。
2.5 如何选择通信方式？
很多新手会问：什么时候用话题，什么时候用服务，什么时候用动作？我总结了一个简单的判断逻辑：
- 话题：数据流持续不断，比如传感器数据、状态信息。发布者不知道谁在收，订阅者不知道谁在发。
- 服务：一次性的请求-应答，比如查询参数、触发某个操作。需要等待结果。
- 动作：长时间执行的任务，需要进度反馈，可能被取消。比如导航、机械臂运动。
我的经验：实际项目中，80%的通信用话题就够了。服务和动作只在特定场景使用。别为了炫技而过度设计，简单就是最好的。
嗯，这一章的内容就到这。核心概念理解了，后面写代码就会顺手很多。记住：节点是基础，话题是主力，服务是补充，动作是利器。下一章我们开始实战，用这些概念搭一个完整的机器人系统。