2026-09-05

rclpy 源码详细分析

工作区路径：/home/cp/work2/ros2Learn/ros2_humble/src/ros2/rclpy
版本：3.3.21（Humble），单包仓库，构建类型 ament_cmake（Python + C++ 扩展），许可证 Apache 2.0。

rclpy 是 ROS 2 Python 客户端库，为 ros2cli、Launch 测试节点、大量 Python 应用提供主 API。它在 rcl（C API）之上通过 pybind11 扩展 _rclpy_pybind11 封装底层句柄，在纯 Python 层实现 Executor、callback_group、参数服务、async 回调 等机制。与 rclcpp 功能对齐，但架构为「Python 调度 + C 绑定」双层设计。

1. 总体认识

1.1 核心职责

能力	说明
生命周期	`rclpy.init()` / `rclpy.shutdown()`，`Context` 封装 `rcl_init`/`rcl_shutdown`
Node	创建 pub/sub/service/client/timer，维护实体列表
Executor	基于 `rcl_wait_set` 调度 ready 回调，支持 sync/async 协程
CallbackGroup	MutuallyExclusive / Reentrant 并发控制
参数	declare/get/set + `ParameterService`（`rcl_interfaces` srv）
时间	`Clock` / `TimeSource`，订阅 `/clock` 实现 sim time
QoS	Python 封装 `rmw_qos_profile_t`，支持 qos_overrides
Action	`rclpy.action` 子模块，封装 `rcl_action`
Lifecycle	`rclpy.lifecycle` 子模块，封装 `rcl_lifecycle`

1.2 在 ROS 2 栈中的位置

对比项	rclpy	rclcpp
语言绑定	pybind11 → Python	原生 C++
Executor 回调	`Task` + `async`/`await` 支持	直接函数调用
Intra-process	Humble 未实现	`IntraProcessManager`
Action/Lifecycle	同仓库 Python 子模块	独立子包
句柄生命周期	`Destroyable` 上下文管理器	RAII + shared_ptr

与 rcl 的分工：与 rclcpp 相同，参数服务实现主体在 rclpy；rcl 负责 CLI/YAML 参数解析。Executor、callback_group 为 rclpy 独有 Python 层逻辑。

2. 仓库结构

rclpy/
├── rclpy/                          # 主包 ★
│   ├── rclpy/                      # Python 源码（~50 个 .py）
│   │   ├── node.py                 # 最大（~1957 行）
│   │   ├── executors.py            # Executor（~870 行）
│   │   ├── qos.py                  # QoS（~499 行）
│   │   ├── action/                 # Action client/server
│   │   ├── lifecycle/              # LifecycleNode
│   │   └── impl/                   # C 扩展懒加载
│   ├── src/rclpy/                  # C++ 绑定（30 个 .cpp）
│   │   ├── _rclpy_pybind11.cpp     # 模块入口
│   │   ├── wait_set.cpp            # rcl_wait_set 封装
│   │   ├── node.cpp / publisher.cpp / subscription.cpp ...
│   │   └── action_*.cpp / lifecycle.cpp
│   ├── test/                       # pytest + gtest
│   └── CMakeLists.txt              # 构建 Python 包 + 扩展
└── (无独立子 package.xml)

2.1 源码规模

层级	文件数	规模
Python（`rclpy/rclpy/`）	~50	最大单文件 `node.py`（1957 行）
C++ 扩展（`src/rclpy/`）	30	最大单文件 `node.cpp`（584 行）、`signal_handler.cpp`（641 行）
Python + C++ 合计	—	~14K 行（不含测试）

3. 依赖关系（`package.xml`）

rclpy
├── rcl                       # C 客户端库
├── rcl_action                # Action C API（扩展层直接调用）
├── rcl_lifecycle             # Lifecycle C API
├── rcl_yaml_param_parser     # YAML 参数（经 Node C++ 层）
├── rcl_logging_interface     # 日志接口
├── rmw / rmw_implementation # 中间件
├── rcutils / rcpputils       # 工具库
├── rosidl_runtime_c          # 消息 C 运行时（扩展层序列化）
├── pybind11_vendor           # Python 绑定
├── rcl_interfaces            # 参数 srv/msg（exec_depend）
├── rosgraph_msgs             # /clock
├── builtin_interfaces        # Time 等
├── unique_identifier_msgs    # Action goal UUID
└── rpyutils                  # import_c_library 工具

4. 双层架构：Python ↔ `_rclpy_pybind11`

4.1 懒加载 C 扩展

为避免 import rclpy 时立即加载 C 库，扩展通过单例延迟导入：

from rpyutils import import_c_library
package = 'rclpy'

rclpy_implementation = import_c_library('._rclpy_pybind11', package)

各模块在需要时 from rclpy.impl.implementation_singleton import rclpy_implementation as _rclpy。

4.2 pybind11 模块导出

_rclpy_pybind11.cpp 注册所有 C 层类型与自由函数：

类别	绑定内容
生命周期	`Context`
Node	`Node`（`rcl_node_t`）
通信	`Publisher`, `Subscription`, `Service`, `Client`, `Timer`
同步	`GuardCondition`, `WaitSet`, `Clock`, `Duration`
Action	`ActionClient`, `ActionServer`, `ActionGoalHandle`
Lifecycle	lifecycle 状态机 C 封装
Graph	`rclpy_get_topic_names_and_types` 等
工具	topic/namespace 校验、remap、序列化、QoS 兼容性检查
异常	`RCLError`, `InvalidHandle`, `TimerCancelledError` 等

4.3 Destroyable：句柄生命周期

C++ 层所有 rcl 句柄继承 Destroyable，实现 Python 上下文管理器：

class Destroyable
{
public:
  void enter();    // __enter__ — 阻止销毁
  void exit(...);  // __exit__ — 允许销毁
  void destroy_when_not_in_use();
  virtual void destroy() = 0;
private:
  size_t use_count = 0u;
  bool please_destroy_ = false;
};

Python 侧统一使用 with handle: 保护 rcl 调用，避免 wait 期间句柄被 GC 销毁：

1 2	with self.handle: self.__publisher.publish(msg)

5. Context 与 init

5.1 Python Context

class Context:
    def init(self, args=None, *, initialize_logging=True, domain_id=None):
        from rclpy.impl.implementation_singleton import rclpy_implementation as _rclpy
        ...
        self.__context = _rclpy.Context(
            args if args is not None else sys.argv,
            domain_id if domain_id is not None else _rclpy.RCL_DEFAULT_DOMAIN_ID)
        if initialize_logging and not self._logging_initialized:
            _rclpy.rclpy_logging_configure(self.__context)

5.2 C++ Context → rcl_init

Context::Context(py::list pyargs, size_t domain_id)
{
  rcl_context_ = std::shared_ptr<rcl_context_t>(...);
  *rcl_context_ = rcl_get_zero_initialized_context();
  // rcl_init_options + rcl_init ...
}

全局 g_contexts 向量跟踪所有有效 context，进程退出时 shutdown_contexts() 统一 shutdown。

5.3 便捷 API（`rclpy/init.py` + `utilities.py`）

API	作用
`rclpy.init()`	默认 context init + 安装信号处理器
`rclpy.ok()`	检查 context 是否有效
`rclpy.shutdown()`	shutdown + 销毁全局 Executor
`rclpy.spin(node)`	全局 SingleThreadedExecutor
`get_default_context()`	进程级 context 单例
`remove_ros_args()`	剥离 ROS CLI 参数

6. Node

6.1 实体容器

Node 在 Python 层维护所有通信实体的列表（rclcpp 则分散在 node_interfaces 中）：

self._publishers: List[Publisher] = []
self._subscriptions: List[Subscription] = []
self._clients: List[Client] = []
self._services: List[Service] = []
self._timers: List[Timer] = []
self._guards: List[GuardCondition] = []
self.__waitables: List[Waitable] = []
self._default_callback_group = MutuallyExclusiveCallbackGroup()

底层 rcl_node_t 由 C++ rclpy::Node 持有；Python Node 通过 property 暴露 handle。

6.2 构造流程

校验 context 已 init
_rclpy.Node(name, namespace, context, cli_args, use_global_arguments, enable_rosout)
创建 logger、TimeSource、ParameterService（可选）
处理 parameter_overrides、qos_overrides 声明

6.3 create_* 方法

方法	Python 包装	C 绑定
`create_publisher`	`Publisher`	`rcl_publisher_init`
`create_subscription`	`Subscription`	`rcl_subscription_init`
`create_service`	`Service`	`rcl_service_init`
`create_client`	`Client`	`rcl_client_init`
`create_timer`	`Timer`	`rcl_timer_init`
`create_guard_condition`	`GuardCondition`	`rcl_guard_condition_init`

创建时实体加入指定 callback_group（默认 MutuallyExclusiveCallbackGroup），并 append 到 Node 对应列表。

7. Publisher / Subscription

7.1 Publisher

Python 薄包装，核心 publish 在 C++：

def publish(self, msg: Union[MsgType, bytes]) -> None:
    with self.handle:
        if isinstance(msg, self.msg_type):
            self.__publisher.publish(msg)
        elif isinstance(msg, bytes):
            self.__publisher.publish_raw(msg)

C++ 层将 Python 消息对象转为 C 结构后调用 rcl_publish：

void Publisher::publish(py::object pymsg)
{
  auto raw_ros_message = convert_from_py(pymsg);
  rcl_ret_t ret = rcl_publish(rcl_publisher_.get(), raw_ros_message.get(), NULL);
  ...
}

消息转换依赖 rosidl_generator_py 生成的 C 绑定与 common_get_type_support() 查找 typesupport。

7.2 Subscription

take_message() 在 C++ 层完成（rcl_take + Python 对象构造）
支持 raw=True 返回 bytes
_executor_event 标志防止同一 subscription 被重复加入 wait set

7.3 数据路径

1 2	publish: Python msg → convert_from_py → rcl_publish → rmw → DDS take: DDS → rmw → rcl_take → Python msg → user callback

注意：Humble 版 rclpy 无 intra-process 优化，所有消息均走完整 rcl/rmw 路径。同进程 Python 节点间通信仍经 DDS（与 rclcpp intra-process 不同）。

8. Executor 架构

Executor 是 rclpy 最复杂的纯 Python 模块（870 行），负责 wait → take → execute 全流程。

8.1 类层次

类	特点
`Executor`	基类：wait set 构建、Task 调度、实体管理
`SingleThreadedExecutor`	调用线程同步执行 `handler()`
`MultiThreadedExecutor`	`ThreadPoolExecutor` 并行提交 handler

8.2 spin 流程

# SingleThreadedExecutor
while rclpy.ok() and not self._is_shutdown:
    handler, entity, node = wait_for_ready_callbacks(timeout)
    handler()  # 同步执行 Task

1
2
3

# MultiThreadedExecutor
handler, entity, node = wait_for_ready_callbacks(...)
self._executor.submit(handler)  # 线程池异步执行

8.3 _wait_for_ready_callbacks 核心逻辑

关键步骤（executors.py）：

遍历节点，收集 can_execute 过滤后的 subscriptions/timers/clients/services/guards/waitables
用 _rclpy.WaitSet 封装 rcl_wait_set_init/clear/add/wait
wait_set.wait(timeout_nsec) 阻塞
按 ready 索引匹配实体，调用 _make_handler 生成 Task
通过 generator yield 返回 (handler, entity, node)

8.4 Task 与 async 回调

rclpy 独有：回调可以是 coroutine。

async def await_or_execute(callback, *args):
    if inspect.iscoroutinefunction(callback):
        return await callback(*args)
    else:
        return callback(*args)

_make_handler 创建 async Task，流程：

callback_group.beginning_execution(entity) — MutuallyExclusive 加锁
take_from_wait_list(entity) — take 消息/请求
await call_coroutine(entity, arg) — 执行用户回调
callback_group.ending_execution(entity) — 释放锁
触发 guard condition 唤醒 wait

8.5 WaitSet C++ 封装

WaitSet::WaitSet(..., Context & context)
{
  rcl_wait_set_init(..., context.rcl_ptr(), ...);
}
// add_subscription / add_timer / add_service / add_client / add_guard_condition
// wait() → rcl_wait()
// get_ready_entities() → 返回 ready 句柄 pointer 集合

与 rclcpp Executor::wait_for_work 直接对应，但 rclpy 在 Python 层 构建 wait set（每轮循环重建），而非 rclcpp 的 MemoryStrategy。

9. CallbackGroup 并发模型

1 2	# ReentrantCallbackGroup — can_execute 恒 True # MutuallyExclusiveCallbackGroup — _active_entity 锁，同时仅一个回调

类型	实现
MutuallyExclusiveCallbackGroup	`threading.Lock` + `_active_entity`，默认 group
ReentrantCallbackGroup	无限制，用于 `Rate` 等

与 rclcpp 的 can_be_taken_from 原子标志语义等价，但 rclpy 用 Python Lock 实现。

Executor.can_execute 额外检查 entity._executor_event：已生成 handler 但未执行完毕的实体不再进入 wait set。

10. 参数系统

10.1 Node 内参数存储

_parameters: dict — 参数名 → Parameter
_descriptors — 参数描述符
_parameters_callbacks — on_parameters_set 链
支持 allow_undeclared_parameters、automatically_declare_parameters_from_overrides

10.2 ParameterService

class ParameterService:
    def __init__(self, node):
        node.create_service(DescribeParameters, nodename + '/describe_parameters', ...)
        node.create_service(GetParameters, nodename + '/get_parameters', ...)
        node.create_service(GetParameterTypes, ...)
        node.create_service(ListParameters, ...)
        node.create_service(SetParameters, ...)
        node.create_service(SetParametersAtomically, ...)

服务名格式为 <node_name>/<service_suffix>（与 rclcpp 的 FQN 路径等价，表达方式不同）。类型均来自 rcl_interfaces（参见 rcl_interfaces 分析）。

10.3 TimeSource 与 use_sim_time

TimeSource 订阅 /clock，监听 use_sim_time 参数变化，更新关联 ROSClock：

def _subscribe_to_clock_topic(self):
    self._clock_sub = node.create_subscription(
        rosgraph_msgs.msg.Clock, CLOCK_TOPIC,
        self.clock_callback,
        QoSProfile(depth=1, reliability=ReliabilityPolicy.BEST_EFFORT))

11. QoS

qos.py（499 行）提供：

QoSProfile 类（depth、reliability、durability、history、liveliness 等）
预设：qos_profile_sensor_data、qos_profile_parameters、qos_profile_services_default 等
qos_overriding_options.py — 与 rclcpp 对齐的运行时 QoS 覆盖

C 层 rclpy_qos_check_compatible() 暴露 RMW 兼容性检查。

12. Action（`rclpy.action`）

同仓库子模块，非独立 package。

文件	职责
`action/client.py`	`ActionClient`，继承 `Waitable`，goal send/cancel/result
`action/server.py`	`ActionServer`，goal 接受与执行
`action/graph.py`	action 图 introspection

C 绑定：action_client.cpp（322 行）、action_server.cpp（430 行），直接调用 rcl_action_*。

Action client 作为 Waitable 注册到 Executor，与 rclcpp action client 设计一致。

13. Lifecycle（`rclpy.lifecycle`）

文件	职责
`lifecycle/node.py`	`LifecycleNode`（多继承 `Node` + `LifecycleNodeMixin`）
`lifecycle/publisher.py`	`LifecyclePublisher` — Active 状态才 publish
`lifecycle/managed_entity.py`	托管实体基类

C 绑定：lifecycle.cpp（364 行）封装 rcl_lifecycle 状态机。

状态转移回调：on_configure、on_activate、on_deactivate、on_cleanup、on_shutdown 等。

14. 其他重要模块

模块	职责
`task.py`	`Future` / `Task` — async 回调与 done 链
`waitable.py`	Waitable 基类，action client 等扩展点
`guard_condition.py`	用户自定义唤醒条件
`signals.py`	SIGINT/SIGTERM → shutdown context
`logging.py`	Python logging 与 rcutils 桥接
`serialization.py`	消息序列化/反序列化
`type_support.py`	消息/服务类型校验
`clock.py` / `time.py` / `duration.py`	时间抽象
`qos_event.py`	QoS 事件（incompatible QoS 等）
`client.py` / `service.py`	请求-响应 RPC
`timer.py`	定时器 + `Rate`
`wait_for_message.py`	阻塞等待首条消息
`validate_*.py`	名称校验纯 Python 实现

15. 关键数据路径

15.1 订阅回调端到端

15.2 典型程序结构

import rclpy
from rclpy.node import Node

def main():
    rclpy.init()
    node = Node('my_node')
    pub = node.create_publisher(String, 'topic', 10)
    sub = node.create_subscription(String, 'topic', callback, 10)
    rclpy.spin(node)   # 全局 SingleThreadedExecutor
    node.destroy_node()
    rclpy.shutdown()

等价于 rclcpp 的 init → Node → spin → shutdown，但 rclpy.spin 使用模块级全局 Executor（__init__.py 中 get_global_executor()）。

16. C++ 扩展文件索引

文件	行数	职责
`node.cpp`	584	Node 创建、参数 YAML、graph
`signal_handler.cpp`	641	信号处理
`action_server.cpp`	430	Action server 绑定
`utils.cpp`	368	消息转换、通用工具
`lifecycle.cpp`	364	Lifecycle 状态机
`action_client.cpp`	322	Action client 绑定
`wait_set.cpp`	295	Wait set
`graph.cpp`	280	图 introspection
`publisher.cpp`	192	publish / publish_raw
`subscription.cpp`	—	take_message
`context.cpp`	188	rcl_init/shutdown
`_rclpy_pybind11.cpp`	241	模块注册入口

17. 与 rclcpp 对照

功能	rclcpp	rclpy
Node 组织	node_interfaces 组合	单类 + 实体列表
Executor wait	C++ MemoryStrategy + rcl_wait	Python 构建 WaitSet + rcl_wait
回调执行	同步函数	Task + 可选 async
句柄保护	shared_ptr RAII	Destroyable + `with handle`
Intra-process	有	无（Humble）
参数服务	ParameterService C++	ParameterService Python
Composable Node	rclcpp_components	无等价物
Static Executor	有	无

两者均调用相同 rcl_* API，调试通信问题时应沿 rcl → rmw 链向下追踪。

18. 调试与测试

日志：RCLCPP_* 对应 rclpy 的 node.get_logger().info() 等
异常：C 层错误转为 RCLError/RMWError/InvalidHandle
测试：test/ 下 ~40 个 pytest 文件 + C++ gtest（test_python_allocator.cpp）
常见问题：
- NotInitializedException — 未调用 rclpy.init()
- 回调不触发 — 未 spin 或实体不在 Executor 管理的节点中
- InvalidHandle — 在 with handle 外使用已销毁句柄
- async 回调在 SingleThreadedExecutor 中需 Task 驱动（不支持裸 event loop）

19. 推荐阅读顺序

impl/implementation_singleton.py + _rclpy_pybind11.cpp — 理解绑定入口
context.py + context.cpp — init/shutdown
node.py（构造 + create_publisher/subscription） — 用户 API
publisher.py + publisher.cpp — publish 数据路径
executors.py（_wait_for_ready_callbacks + _make_handler） — 调度核心
callback_groups.py — 并发语义
wait_set.cpp — 底层 wait
parameter_service.py + parameter.py — 参数栈
time_source.py — sim time
action/client.py + action_client.cpp — Action 与 Waitable
lifecycle/node.py — 生命周期
对照 rcl 源码分析、rclcpp 源码分析

20. 小结

rclpy 是 ROS 2 Python 应用的主 API 层，采用 Python 调度 + pybind11 C 绑定 双层架构：

Python 层（node.py、executors.py）负责 Entity 管理、Executor 调度、async 回调、参数服务
C 扩展（_rclpy_pybind11）负责 rcl_* 调用、消息转换、WaitSet、Action/Lifecycle 绑定
Destroyable + with handle 保证多线程 wait 期间句柄安全
Task/async 是 rclpy 相对 rclcpp 的显著差异，支持 coroutine 风格节点

掌握 _wait_for_ready_callbacks → _make_handler → Task 执行 链路，是理解 rclpy 并发模型的关键。

文章互动

阅读 --

留言

0 条留言

正在加载留言…