ROS 2 里那个"第一个节点",代码抄下来也就十几行,可它背后串着环境变量、构建系统、Python 解释器、执行器、DDS 发现机制一整套东西,新手上手翻车的概率其实相当高。我见过太多人colcon build成功、ros2 run敲下去却什么也不打印,或者 Ctrl+C 之后进程赖在那里不走。这篇就围绕《ROS 2机器人开发从入门到实践》里 2.1.1 这一节的内容,把"写一个 Python 节点"从版本搭配、功能包结构、代码逐行拆解,一直讲到跑起来之后怎么验证、卡住之后怎么查。适合刚装完 ROS 2、手上只有一台 Ubuntu 机器、想先跑通一个能打印日志的最小节点的同学;如果你已经能写话题订阅了,里面关于参数声明、执行器和退出流程的那几段,也值得翻一翻。
1. 先把"节点"这件事说透:第一个例子为什么要这么设计
1.1 节点是 ROS 2 的最小运行单元,但它不只是"一个进程"
在 ROS 2 的语境里,节点(Node)是参与计算的最小单元。它通常对应一个操作系统进程,进程内部可以跑若干发布者、订阅者、服务端、客户端、定时器。跟 ROS 1 最大的区别是:ROS 2 里没有 master 这个中心节点了,每个节点通过底层 DDS 自己做发现(discovery),谁发布了什么话题、谁提供了什么服务,靠节点之间互相打招呼来交换信息。这件事对写代码的影响是零,但对排查问题的影响非常大——以前"master 没起来"是一个明确的分诊点,现在你得靠ros2 node list、ros2 topic list这些工具自己看图。
节点名有硬性约束:只能用字母、数字、下划线,不能以数字开头,不能带空格和连字符。这个名字在ros2 node list里就是它的身份标识。名字重了不会报错,两张图混在一起时排查会非常难受,所以我一般在实验阶段就用命名空间或者__node:=重映射把它区分开,别等到出问题再回头改。
还有一个容易忽略的点:节点构造的时候就会触发 DDS 参与者(participant)的创建,这个过程要申请网络资源、建立发现通道。所以"创建节点"不是一件轻量的事,一个进程里塞几百个节点,启动时间和资源占用都会明显变差。后面讲组件化(composable node)就是为了解决这个问题,不过在 2.1.1 这个阶段,一个进程一个节点完全够用。
1.2 第一课选 Python,是因为反馈环路最短
很多人问第一门语言该用 C++ 还是 Python。我的答案很直接:验证概念用 Python,追求控制周期用 C++。
Python 的rclpy已经覆盖了 ROS 2 的核心能力——话题、服务、动作(Humble 里 rclpy 的动作支持已经完整)、参数、定时器、日志,一个都不缺。它的优势是改完源码不用编译,直接重新运行就能看到效果;配合--symlink-install,大部分改动连 build 都省了。这个反馈速度,在你调试"回调到底有没有被触发""参数有没有读到"这类问题时,价值极大。
它的短板也很明确:全局解释器锁加上解释执行的开销,让 Python 很难稳定跑在几百赫兹以上的硬实时控制回路里。我的经验值是这样——周期性任务在 20 到 50 赫兹这个量级,Python 完全没问题;上到几百赫兹且抖动敏感,就该考虑 C++ 或者把这段逻辑放到微控制器上。传感器数据桥接、状态机编排、调试脚本、参数管理这些东西,用 Python 写维护成本最低,没必要硬上 C++。
顺便说一句,现在不少人拿 ros 2 humble 搭配 micro-ROS 跑在 ESP32 这类小板子上,玩法很有代表性:板子端跑实时性要求高的部分,上位机用 Python 节点做数据汇聚和可视化。这个组合里,Python 节点的稳定性往往比板子端更值得关注。
1.3 一个能跑的最小节点,骨架就是四块
不管代码写多少行,任何一个rclpy节点都逃不出这四步:初始化上下文、创建节点对象、把控制权交给执行器、退出时清理资源。
拿生活里的场景类比一下。rclpy.init()相当于给办公室通电、把网络和电话线接好,这一步会创建 DDS 参与者、解析命令行参数(比如--ros-args后面的东西)、初始化日志系统;创建节点对象相当于招聘一个员工并给他发工牌,工牌上写的就是节点名;rclpy.spin(node)相当于让这个员工坐在工位上值班,有人打电话(回调触发)就处理;destroy_node()加rclpy.shutdown()则是下班交接、关电闸。
顺序不能乱。没 init 就建节点,会直接在构造时报"context is not initialized";spin 之后忘了 destroy,资源释放不干净,短时间反复重启同一个节点,偶尔会遇到端口占用或者发现异常之类的怪问题。这四步看着啰嗦,但写顺手之后就变成本能了。
2. 环境准备:把 ROS 2 和 Python 对上号,别在第一步埋雷
2.1 版本怎么搭:Humble 配 Ubuntu 22.04 配 Python 3.10
这三个东西是绑定的。ROS 2 Humble 官方支持 Ubuntu 22.04,而 22.04 系统自带的 Python 就是 3.10,rclpy在编译时也链接到系统这个解释器。你要是换成 Ubuntu 24.04 想装 Humble,或者反过来在 22.04 上装 Jazzy,各种依赖冲突会把你拖进坑里,纯属自找麻烦。
动手前先做三件确认:
lsb_release -a python3 --version which python3 echo $ROS_DISTRO第一条看系统版本,要能看到22.04;第二条应该是Python 3.10.x;第三条最关键,which python3输出的必须是/usr/bin/python3,如果它指到/home/xxx/miniconda3/bin/python3或者别的路径,后面一定会出ModuleNotFoundError: No module named 'rclpy'。原因是rclpy是编译好的二进制模块,装在系统 Python 的 site-packages 里,你换一个解释器,它就找不到了。最后一条的输出应该是humble,空的就说明没 source 环境。
2.2 安装完成后的三件事:source、验证、隔离
装完 ROS 2,第一件事是让当前终端认识它:
source /opt/ros/humble/setup.bash然后把它写进~/.bashrc,省得每开一个终端都要敲一遍:
echo "source /opt/ros/humble/setup.bash" >> ~/.bashrc第二件事是跑官方示例验证整条链路。开两个终端,一个跑:
ros2 run demo_nodes_py talker另一个跑:
ros2 run demo_nodes_py listener看到 listener 端不停打印I heard: [Hello World: 23]这类信息,说明 Python 节点、DDS 发现、话题通信全部正常。这个验证动作我建议形成肌肉记忆,任何"节点跑不起来"的问题,都先用它确认是环境坏了还是自己代码坏了。
第三件事是隔离。如果你平时用 conda 管理 Python 环境,跑 ROS 2 之前先conda deactivate。如果你的~/.bashrc里有自己加的PYTHONPATH,也要留意,它可能让 Python 优先加载到你本地的某个同名包,产生一些莫名其妙的导入错误。
注意:不要用 pip 去装 rclpy。它是一个跟特定 ROS 发行版版本绑定的包,从 PyPI 上装的那份跟系统里的 DDS 实现未必对得上,出问题极难排查。
2.3 创建功能包:目录结构和构建类型的选择
ROS 2 里的代码必须放在"功能包"(package)里,ros2 run才能找到它。Python 包用ament_python构建类型:
mkdir -p ~/ros2_ws/src cd ~/ros2_ws/src ros2 pkg create --build-type ament_python --node-name hello_node my_first_pkg这几行会生成这样一棵树:
my_first_pkg/ ├── my_first_pkg/ │ ├── __init__.py │ └── hello_node.py ├── package.xml ├── setup.py ├── setup.cfg └── resource/ └── my_first_pkg几个位置的作用要讲清楚。外层的my_first_pkg/是包目录,setup.py和package.xml是它的"身份证",构建系统靠这两个文件认识它。内层的my_first_pkg/是真正的 Python 模块目录,你的节点代码就放这里。resource/my_first_pkg是一个空文件,作用是给 ament 索引做标记,ros2 run、ros2 pkg list能找到这个包全靠它——我见过有人觉得这文件没用顺手删了,结果ros2 run报"Package not found",找了大半天。
--node-name hello_node这个参数会顺手生成一个可运行的模板节点,并且自动把它写进setup.py的入口点。刚开始学的时候用它很省事,能直接看到"能跑起来的样子"。
构建命令:
cd ~/ros2_ws colcon build --symlink-install --packages-select my_first_pkg source install/setup.bash--packages-select只构建指定的包,工作空间里有几十个包的时候能省不少时间。--symlink-install会让安装目录里的 Python 文件变成指向源码的符号链接,好处是你改了.py文件之后不用重新 build 就能生效。但有个例外要记住:改了setup.py里的entry_points(比如新增了节点名),必须重新 build,否则ros2 run找不到新名字。
还有一点,构建完之后必须source install/setup.bash,而且要在当前这个终端里。很多人开新终端只 source 了/opt/ros/humble/setup.bash,忘了 source 工作空间,于是ros2 run一直报找不到包。这个坑可以说是新手第一大坑。
3. 从零写第一个节点:代码逐行拆开看
3.1 最简版本:先确认环境是通的
我们不用模板,手写一个最小版本,存成my_first_pkg/my_first_pkg/hello_node.py:
import rclpy from rclpy.node import Node def main(args=None): rclpy.init(args=args) # 1. 初始化上下文 node = Node('hello_node') # 2. 创建节点,名字叫 hello_node node.get_logger().info('你好,ROS 2') # 3. 打一条日志 rclpy.spin(node) # 4. 进入事件循环,等待回调 node.destroy_node() # 5. 释放节点资源 rclpy.shutdown() # 6. 关闭上下文 if __name__ == '__main__': main()逐行说。rclpy.init(args=args)里的args是从命令行传进来的参数列表,默认是None的话它就去读sys.argv。这一行会初始化上下文(Context),创建 DDS 参与者,解析--ros-args后面的重映射和参数覆盖。你在命令行里写的那些参数能生效,全靠这一步。
Node('hello_node')直接实例化基类,得到一个只有名字、没有任何接口的节点。这种做法在教学里很常见,实际项目里几乎不会这么写,因为节点总归要做点事,一般会继承Node写一个自己的类。get_logger()返回的是 ROS 2 的日志器,它跟 Python 原生的 logging 是打通的,打印格式里会带时间戳、日志级别、节点名,比print强得多,而且支持运行时按级别过滤。新写代码我强烈建议一律用get_logger(),不要用print——print的输出不进入 ROS 日志体系,ros2 launch起来之后你可能根本看不到它。
rclpy.spin(node)是关键。它是一个阻塞调用,会把当前线程交给执行器,让它不停处理这个节点上的回调。如果节点没有任何定时器、订阅者,它就会安静地卡在那里等 Ctrl+C。这也解释了为什么很多人觉得"运行之后什么都没发生"——没有回调,自然没有输出。
最后destroy_node()和shutdown(),在 ROS 2 里做清理。跑一下:
cd ~/ros2_ws colcon build --symlink-install --packages-select my_first_pkg source install/setup.bash ros2 run my_first_pkg hello_node看到[INFO] [1750000000.123456789] [hello_node]: 你好,ROS 2这行输出,环境就算是打通了。
3.2 定时器版本:让节点持续干活
只会打印一行日志的节点没什么用。我们把上面的代码改造成一个带定时器的类,让它每秒输出一次心跳,存成counter_node.py:
import rclpy from rclpy.node import Node class CounterNode(Node): def __init__(self): super().__init__('counter_node') self.count = 0 self.period = 1.0 self.timer = self.create_timer(self.period, self.on_timer) self.get_logger().info(f'节点已启动,周期 {self.period} 秒') def on_timer(self): self.count += 1 self.get_logger().info(f'心跳计数:{self.count}') def main(args=None): rclpy.init(args=args) node = CounterNode() try: rclpy.spin(node) except KeyboardInterrupt: pass finally: node.destroy_node() if rclpy.ok(): rclpy.shutdown() if __name__ == '__main__': main()这里有几个点值得展开。super().__init__('counter_node')必须放在最前面,因为后面所有create_*调用都依赖基类初始化完成,顺序错了会报"node is not initialized"之类的问题。
create_timer(period, callback)创建的定时器由执行器统一管理,period 单位是秒,可以是浮点数,0.05 就是 20 赫兹。回调函数是在执行器线程里被调用的,所以回调里做的活越重,定时精度就越差,而且会拖慢同一执行器上的其他回调。这是个很关键的性能认知:单线程执行器下,所有回调排队执行,一个慢回调会堵住整个节点。
try / except KeyboardInterrupt / finally这套结构不是装饰,是必须的。rclpy.spin阻塞在主线程,Ctrl+C 会抛KeyboardInterrupt,如果你不接住它,destroy_node和shutdown就不会被执行,节点结束得不干净。这在短时间反复启动同一个节点做测试时特别明显。
运行:
ros2 run my_first_pkg counter_node另一个终端里执行ros2 node list,应该能看到/counter_node。
3.3 参数版本:把写死的数值变成运行时可调
self.period = 1.0这种硬编码在调试阶段非常难受,每改一次周期都要动源码、重新跑。ROS 2 的参数系统就是解决这个问题的:
class CounterNode(Node): def __init__(self): super().__init__('counter_node') self.declare_parameter('period', 1.0) # 声明参数,默认值 1.0 秒 self.declare_parameter('node_label', '心跳') # 字符串参数 period = self.get_parameter('period').get_parameter_value().double_value label = self.get_parameter('node_label').get_parameter_value().string_value self.count = 0 self.label = label self.timer = self.create_timer(period, self.on_timer) self.get_logger().info(f'启动完成,周期 {period} 秒')declare_parameter(name, default_value)里的默认值决定了参数类型:写1.0就是 double,写1就是 integer,写'abc'就是 string。这个类型一旦确定,后面用别的方式覆盖它时会做类型检查。取值用get_parameter(...).get_parameter_value().double_value,也可以用简写get_parameter(...).value,后者在类型明确时更好读。
有个坑必须提醒:同名参数声明两次会直接抛ParameterAlreadyDeclaredException。在继承体系里,如果父类声明过某个参数,子类不要重复声明,用has_parameter()判断一下更稳妥。
命令行覆盖参数:
ros2 run my_first_pkg counter_node --ros-args -p period:=0.5再提醒一个实测踩过的坑:如果参数声明的是 double,你在命令行写-p period:=2,因为参数值是按 YAML 解析的,2会被当成整数,类型对不上,会报参数类型不匹配的提示。写成-p period:=2.0就没事。这个细节文档里往往一笔带过,但非常容易卡人。
运行时还能改:
ros2 param list /counter_node ros2 param get /counter_node period ros2 param set /counter_node period 0.2改完之后,节点下的定时器频率会立刻变化,因为set之后会触发参数回调——不过要注意,上面的代码里定时器周期是在__init__里读一次然后固定下来的,param set不会自动重建定时器。想让它真正动态生效,得注册add_on_set_parameters_callback,在回调里重建定时器。这个属于下一层的内容,但知道"参数改了不一定生效"这件事本身,能省下你半小时的困惑。
3.4 打包与运行:setup.py 里那行 entry_points 决定一切
节点写完了,得让ros2 run能找到它。打开setup.py,确认入口点这一节:
from setuptools import setup package_name = 'my_first_pkg' 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'], zip_safe=True, maintainer='you', maintainer_email='you@example.com', description='我的第一个 ROS 2 Python 节点示例', license='Apache-2.0', tests_require=['pytest'], entry_points={ 'console_scripts': [ 'hello_node = my_first_pkg.hello_node:main', 'counter_node = my_first_pkg.counter_node:main', ], }, )entry_points里的格式是可执行名 = 包名.模块名:函数名。左边那串就是你ros2 run后面跟的名字,中间是文件路径(相对内层包目录的模块路径),右边是入口函数。三处任何一处写错,表现都是"命令跑起来没反应"或者"报找不到模块",而且报错信息未必直白。
同时别忘了package.xml里声明依赖,否则换台机器就构建失败:
<exec_depend>rclpy</exec_depend>改完setup.py之后必须重新 build:
cd ~/ros2_ws colcon build --symlink-install --packages-select my_first_pkg source install/setup.bash ros2 run my_first_pkg counter_node如果你只是改了counter_node.py里的逻辑,--symlink-install下直接重跑就行,不用 build。但新增一个节点名、改了入口函数名,就一定要 build。
4. 跑起来之后:验证手段和排查手册
4.1 几个必会命令,够应付九成问题
节点启动之后,ros2 run那个终端就一直占着,得有另一个终端来观察。下面这几条命令我几乎每次调试都用:
| 命令 | 作用 | 实用场景 |
|---|---|---|
ros2 node list | 列出所有在跑的节点 | 确认自己的节点到底起来没有 |
ros2 node info /counter_node | 看节点的接口清单 | 确认订阅、发布、服务、参数有没有挂上 |
ros2 param list /counter_node | 列出节点参数 | 检查 declare 是否成功 |
ros2 param get /counter_node period | 读参数当前值 | 确认命令行覆盖有没有生效 |
ros2 param set /counter_node period 0.2 | 运行时改参数 | 免重启调参 |
ros2 param dump /counter_node | 导出参数为 YAML | 复现问题现场 |
rqt_graph | 图形化看节点拓扑 | 排查"谁连不上谁" |
ros2 param dump这个我特别推荐。它会把当前节点的所有参数写成一个 YAML 文件,下次可以用ros2 run my_first_pkg counter_node --ros-args --params-file params.yaml一键还原整套配置。调参调到某个组合效果好,先 dump 一份存着,比记在便签上靠谱得多。
4.2 典型报错对照表
下面这些是我和身边人真踩过的,按出现频率排:
| 报错或现象 | 大概率原因 | 处理方式 |
|---|---|---|
Package 'my_first_pkg' not found | 没 source 工作空间的 setup.bash | 在项目根目录执行source install/setup.bash |
ModuleNotFoundError: No module named 'rclpy' | 用了 conda 或其他 Python 解释器 | conda deactivate,确认which python3指向/usr/bin/python3 |
ros2 run有反应但没有输出 | 入口点名字写错,或回调没被触发 | 检查setup.py的 entry_points,检查定时器/订阅是否创建成功 |
parameter 'period' has invalid type | 命令行传了整数给 double 参数 | 写成-p period:=2.0 |
ParameterAlreadyDeclaredException | 同名参数声明两次 | 用has_parameter()判断后再声明 |
| Ctrl+C 之后进程不退出 | 回调里有阻塞操作,或者异常没接住 | 用 try/finally 包住 spin,检查回调内的耗时操作 |
what(): rcl_shutdown already called | 重复调用 shutdown | 用if rclpy.ok():判断,新版本可用 try_shutdown |
| 节点名重复,行为混乱 | 同名节点同时存在 | 用-r __node:=xxx重命名或加命名空间 |
__node和__ns这两个重映射是 ROS 2 里很实用的技巧,前者改节点名,后者改命名空间:
ros2 run my_first_pkg counter_node --ros-args -r __node:=beat_node ros2 run my_first_pkg counter_node --ros-args -r __ns:=/robot1命名空间加完之后,节点会变成/robot1/counter_node,多台机器人或者同一台机器上跑多份同样的逻辑时特别有用。
4.3 日志级别和 --ros-args 的实用玩法
ROS 2 的日志分五个级别:debug、info、warn、error、fatal。默认只显示 info 及以上,所以你在代码里写的get_logger().debug(...)看不到输出,不是代码错了,是级别不够。运行时调整:
ros2 run my_first_pkg counter_node --ros-args --log-level debug ros2 run my_first_pkg counter_node --ros-args --log-level counter_node:=debug第一条把所有节点的级别都降到 debug,第二条只针对counter_node这一个节点。系统里有十几个节点在跑的时候,第二条明显更好用。反过来,也可以用--log-level error把噪音压下去,只看真正出问题的地方。
日志还有个小技巧:get_logger().info(msg, throttle_duration_sec=1.0)这种节流参数在rclpy里的支持跟 C++ 不太一样,Python 侧更常见的做法是自己写个计数判断,或者干脆把高频日志降级成 debug。高频循环里打 info 级日志,日志系统本身会成为瓶颈,这一点在做几百赫兹的循环时体现得非常明显。
5. 让节点更像个"正经项目":几个实用小改造
5.1 优雅退出:别让 Ctrl+C 留下半死不活的进程
最基础的退出处理在 3.2 已经写了,但还有两个情况要处理。
第一是外部关闭。在 Humble 里,如果上下文被别的代码关掉了,rclpy.spin会抛rclpy.executors.ExternalShutdownException。只接KeyboardInterrupt是接不住的,会打出难看的堆栈。稳妥写法是:
from rclpy.executors import ExternalShutdownException def main(args=None): rclpy.init(args=args) node = CounterNode() try: rclpy.spin(node) except (KeyboardInterrupt, ExternalShutdownException): pass finally: node.destroy_node() if rclpy.ok(): rclpy.shutdown()第二是shutdown被重复调用会抛异常。新一些的发行版提供了rclpy.try_shutdown()来避免这个问题,Humble 上就用if rclpy.ok():判断。这段看起来有点防御式编程的味道,但它在脚本被反复启动和杀掉的时候能省掉大量莫名其妙的报错。
还有一个更隐蔽的问题:回调里做了耗时操作(比如读一个大文件、等一个网络请求),Ctrl+C 之后进程会卡在回调里不出来直到那次操作结束。如果这个操作是无限等待,进程就真的退不出去,只能kill -9。所以我在写回调的时候有个习惯——任何可能阻塞的操作都设超时,或者干脆丢到独立线程里去做。
5.2 执行器与回调阻塞:什么时候该上多线程
rclpy.spin(node)背后用的是全局的单线程执行器。它的行为很好理解:所有回调在一个线程里排队,谁先到谁先执行,前一个没执行完,后面的就得等。这对大多数场景没问题,但只要你的节点同时订阅了多个话题、又有定时器,就可能出现"某个话题卡住了,其他回调全都不响应"的情况。
解决办法是显式使用多线程执行器:
import rclpy from rclpy.executors import MultiThreadedExecutor def main(args=None): rclpy.init(args=args) node = CounterNode() executor = MultiThreadedExecutor(num_threads=2) executor.add_node(node) try: executor.spin() except KeyboardInterrupt: pass finally: executor.remove_node(node) executor.shutdown() node.destroy_node() if rclpy.ok(): rclpy.shutdown()需要提醒的是,多线程不等于问题自动消失。回调之间如果共享了可变状态(比如上面那个self.count),就得自己加锁,或者依赖 Python 的原子性保证来规避竞争——self.count += 1这条语句本身不是原子的,多线程下计数会少。我一般的原则是:能用单线程就用单线程,确实有阻塞源(比如同步调用外部接口)再上多线程,并且把共享状态收拢到一个明确的地方管理。
5.3 从单节点到多节点:命名空间和后续扩展方向
第一个节点跑通之后,下一步一般是加个发布者或者订阅者。在CounterNode里加一个话题发布只要三行:
from std_msgs.msg import String self.publisher = self.create_publisher(String, 'heartbeat', 10) # 在 on_timer 里 msg = String() msg.data = f'count={self.count}' self.publisher.publish(msg)这里的10是队列深度(depth),指的是发布端在网络拥塞时最多缓存多少条消息。默认服务质量(QoS)是可靠传输加 volatile,丢一条都算异常。如果你后面要做的是图像、点云这类高频大流量数据,就得换成 best effort 加小深度的 QoS 配置,否则网络会撑不住。这个话题在 2.1.1 里不展开,但你可以先记住这个位置,等做传感器数据的时候再回来改。
再往后,几个方向值得留意。一是ros2 launch,用 XML 或 Python 描述文件把一堆节点、参数、命名空间一次性拉起来,比手敲ros2 run靠谱得多。二是参数文件,把调试好的参数固化成 YAML,用--params-file加载,团队协作时能保证大家跑的是同一套配置。三是组件化,把一个进程里的多个节点合并到一个进程中执行,减少 DDS 参与者数量和进程间通信开销。Python 侧对组件的支持比 C++ 晚一些,具体可用性跟你的发行版关系很大,动手前先查对应版本的文档,别照着老教程硬套。
我自己在带新人时的体会是,2.1.1 这一节看着最简单,其实最容易留下坏习惯。第一个节点如果就用print打日志、不接异常直接让 spin 裸奔、参数全写死在代码里,后面写几十个节点的时候会一直重复这些毛病。反过来,如果第一份代码就把日志、参数、退出流程写规矩了,后面加订阅加发布只是往框架里填东西。我个人的习惯是把前面那个CounterNode存成模板,新项目复制过来改名字和参数,比每次从空文件开始快得多。