ROS2之节点
什么是节点?
在ROS2(机器人操作系统2)中,节点(node)是执行程序的基本单元,也是构成整个机器人系统的核心“积木”。你可以把它理解为系统中一个独立、可执行的进程,每个节点都专注于完成一个特定的、单一的功能。这种设计哲学让复杂的机器人系统变得模块化,易于开发、维护和扩展。
节点的核心特性
-
模块化(Modularity) 这是节点最重要的特性。每个节点只做一件事,而且做得很好。例如,在一个移动机器人系统中,你不会把所有代码都写在一个庞大的程序里,而是会创建多个独立的节点来完成不同的任务:
-
感知节点:负责从摄像头或激光雷达等传感器收集数据。
-
规划节点:根据感知数据和目标,计算出一条最优路径。
-
控制节点:接收规划结果,并发送指令来控制机器人的电机。 这种分工合作的方式使得开发团队可以并行工作,并且当某个节点出现问题时,不会影响到整个系统的运行。
-
-
分布式(Distributed) 节点可以运行在不同的计算机上,甚至在不同的操作系统上。只要这些设备能通过网络连接,ROS2的通信机制就能让它们无缝地交换信息。例如,一个节点可以在机器人本体上的小型计算机上运行,而另一个负责复杂计算的节点则可以在远程的强大服务器上运行。这为构建复杂的、跨设备的机器人系统提供了极大的灵活性。
-
通信(Communication) 节点之间通过一系列标准的通信方式进行信息交换,这使得它们能够协同工作。最常见的通信方式是:
-
话题(Topics):这是一种发布/订阅(Publish/Subscribe)机制。一个节点(发布者)向一个特定的“话题”发送数据,而对这个话题感兴趣的节点(订阅者)会接收这些数据。例如,一个“激光雷达”节点可以持续向名为
/scan的话题发布扫描数据,而“规划”节点会订阅这个话题来获取数据。 -
服务(Services):这是一种请求/响应(Request/Reply)机制。它类似于传统的函数调用,一个节点向另一个节点发送一个请求,并等待接收一个响应。这通常用于执行一次性的、同步的任务,比如请求机器人执行一个特定的动作。
-
动作(Actions):这是一种用于处理耗时任务的通信方式。一个节点向另一个节点发送一个目标,并能持续接收任务进度反馈,直到任务完成或被取消。
-
使用python代码简单创建节点
import rclpy
from rclpy.node import Node
def main():
rclpy.init() # 初始化
node = Node('node_one') # 创建节点
# 打印节点信息
node.get_logger().info('node_one activate')
node.get_logger().warn('node_one warn')
node.get_logger().error('node_one error')
'''
rclpy.spin的底层是下面三句代码的总和,创建一个Executor即单线程(而非进程)执行器,然后将节点添加到执行器中,开始运行所有节点的回调逻辑,直到shutdown调用最后停止循环
executor = rclpy.executors.SingleThreadedExecutor()
executor.add_node(node)
executor.spin()
在无限循环期间,执行器会不停地:1、检查哪些节点有待处理的事件;2、调用对应的回调函数;3、空闲时进入等待状态(高效阻塞);4、一直到 rclpy.shutdown() 被调用,才停止循环
'''
rclpy.spin(node)
rclpy.shutdown()
if __name__=='__main__':
main()

使用C++代码简单创建节点
首先我们要了解头文件和库文件的关系,因此我们在写代码时导入依赖需要先导入头文件
头文件 提供 声明(告诉编译器有这个函数/类),库文件 提供 实现(真正的函数/类定义)
源代码(main.cpp) + 头文件(.h) --> 编译 --> 目标文件(.o)目标文件(.o) + 库文件(.a/.so) --> 链接 --> 可执行文件
在运行的过程种发现虚拟环境的libstdc++.so.6太落后了,于是更新~/.bashrc加上export LD_LIBRARY_PATH=/usr/lib/x86_64-linux-gnu:$LD_LIBRARY_PATH,优先使用系统库
记得在C++扩展的配置里面加上/opt/ros/humble/include/**的包让C++扩展可以读取到
#include <iostream>
#include <rclcpp/rclcpp.hpp>
using namespace std;
using namespace rclcpp;
int main(int argc,char** argv){
init(argc,argv);
auto node = make_shared<Node >("cpp_node");
RCLCPP_INFO(node->get_logger(),"cpp_node start");
spin(node);
shutdown();
return 0;
}
cmake_minimum_required(VERSION 3.8)
project(helloWorld)
add_executable(hello hello_world.cpp)
find_package(rclcpp REQUIRED)
include_directories(hello PUBLIC ${rclcpp_INCLUDE_DIRS})
target_link_libraries(hello ${rclcpp_LIBRARIES})
mkdir build
cd build
# 会到上一层目录查找CMakeLists.txt文件
cmake ..
make
./hello
关于节点初始化更深入的理解,一般来说由于现实项目的复杂性,我们采用一个程序文件对应一个特定节点的方式,这种方式下每个节点都有自己的进程,init是初始化一个ROS的上下文,对于一个进程来说只有一个ROS上下文,一般也采用创建一个单线程执行器来管理该节点的所有事件,当创建初始化上下文并创建节点之后,这时仅完成事件注册,不处理任何事件,事件会暂存在队列中,之后该进程进入事件循环阻塞,会先处理已经注册的事件,后续像回调函数我们需要通过定时器让其加入事件循环统一调动,这是周期触发的事件,如果我们使用thread创建一个子线程的话,则主线程会一直进行事件循环而子线程可以进行其他的任务,这个任务是不会进入事件循环的,可以独立执行的,会与子线程抢夺CPU,但是子线程也能访问ROS上下文,因为该上下文是进程级别的。
核心结论:节点初始化只会注册事件,真正的执行要等待spin启动之后即给该节点一个执行器处理所有的注册事件,这些注册事件有些会周期触发,因此后续主线程需要一直进行事件循环监听,所以在节点初始化的时候不能阻塞主线程,必须让执行器能够掌握主线程。
功能包
ROS2 中的功能包(package)是组织和管理代码的基本单位。你可以把它看作是一个装有所有相关文件的文件夹,这些文件共同实现了一个或多个特定的功能。
功能包的组成:
-
可执行文件(Nodes):实现具体功能的程序,也就是上面我们提到的节点。
-
库文件(Libraries):可供其他功能包使用的代码库。
-
配置文件:如 YAML 文件,用于设置参数和配置。
-
消息、服务和动作定义(
msg、srv、action):定义了节点之间通信时所使用的标准数据格式。 -
配置文件(
CMakeLists.txt或package.xml):这是功能包的“身份证”和“构建脚本”,package.xml描述了功能包的基本信息和依赖关系,而CMakeLists.txt或其他构建脚本则告诉 ROS2 如何编译和安装这个功能包。 -
其他资源:如启动文件(launch files)、URDF 文件、图片和模型等。
进行简单的功能包的创建,打开终端输入以下命令,创建完功能包后需要先进行一些文件的改动才能进行功能包的构建
ros2 pkg create demo_python_pkg(包名) --build-type ament_python(构建类型) --license Apache-2.0(证书) (创建功能包)
colcon build (构建功能包)
source install/setup.bash (更新环境变量)
echo $AMENT_PREFIX_PATH (输出环境变量)
ros2 run demo_python_pkg python_node _main (执行可执行文件)
在功能包创建完成之后,先创建一个节点文件python_node
import rclpy
from rclpy.node import Node
def main():
rclpy.init()
node = Node("NodeOne")
node.get_logger().info("1号节点启动")
rclpy.spin(node)
rclpy.shutdown()
之后修改setup.py文件和package.xml文件



构建之后得到的可执行文件

如果修改了源码则需要重新构建得到新的可执行文件,注:环境变量的更新只在本终端有效如果新开一个终端需要再次执行一次更新命令
C++实现
cmake_minimum_required(VERSION 3.8)
project(demo_cpp_pkg)
if(CMAKE_COMPILER_IS_GNUCXX OR CMAKE_CXX_COMPILER_ID MATCHES "Clang")
add_compile_options(-Wall -Wextra -Wpedantic)
endif()
# find dependencies
find_package(ament_cmake REQUIRED)
find_package(rclcpp REQUIRED)
# uncomment the following section in order to fill in
# further dependencies manually.
# find_package(<dependency> REQUIRED)
add_executable(cppNode src/cpp_node.cpp)
# include_directories(cppNode PUBLIC ${rclcpp_INCLUDE_DIRS})
# target_link_libraries(cppNode ${rclcpp_LIBRARIES})
# 等同于上面两个命令
ament_target_dependencies(cppNode rclcpp)
# 将可执行文件拷贝到install/lib/demo_cpp_pkg/可执行文件
install(TARGETS cppNode
DESTINATION lib/${PROJECT_NAME}
)
if(BUILD_TESTING)
find_package(ament_lint_auto REQUIRED)
# the following line skips the linter which checks for copyrights
# comment the line when a copyright and license is added to all source files
set(ament_cmake_copyright_FOUND TRUE)
# the following line skips cpplint (only works in a git repo)
# comment the line when this package is in a git repo and when
# a copyright and license is added to all source files
set(ament_cmake_cpplint_FOUND TRUE)
ament_lint_auto_find_test_dependencies()
endif()
ament_package()

ros2 run demo_cpp_pkg cppNode
可看到成功运行,还有pckage.xml不要忘了加依赖
面向对象的思想:将不同的事物分为不同的类,对于每一个类,进行实例化得到对象,同一个的类的对象变量和方法是一样的,但是变量储存的内容可能不一样。
#include <iostream>
#include <rclcpp/rclcpp.hpp>
using namespace std;
using namespace rclcpp;
class CppNode : public Node{
private:
string name;
public:
CppNode(const string &node_name,const string &name):Node(node_name) #需要隐式或者显示调用父类构造函数
{
this->name = name;
};
void myName(){
RCLCPP_INFO(this->get_logger(),"my name is %s",this->name.c_str());
};
};
int main(int argc,char** argv){
init(argc,argv);
auto node = make_shared<CppNode>("cpp_node","bill");
RCLCPP_INFO(node->get_logger(),"cpp_node start");
node->myName();
spin(node);
shutdown();
return 0;
}
argc 表示参数数量,argv 存放参数内容。
在 ROS2 中,工作空间(workspace) 是组织和管理功能包的目录结构,因为一般项目会涉及到多个功能包,常见的标准结构如下:
ros2_ws/ # 工作空间根目录(通常叫 xxx_ws)
├── src/ # 源代码目录,存放功能包
│ ├── my_robot/ # 一个 C++ 或 Python 功能包
│ ├── my_robot_interfaces/ # 一个接口包(存放 msg、srv、action)
│ └── ...
├── install/ # 编译后安装目录,存放可执行文件、库、资源
├── build/ # 编译过程中的中间文件(CMake/colcon 生成)
└── log/ # 构建、运行时的日志文件
src/
你写的所有 ROS2 包都放这里。
可以同时放 Python 包(有 setup.py)和 C++ 包(有 CMakeLists.txt 和 package.xml)。
build/
colcon build 时生成的临时构建文件,类似于 CMake 的 build 目录。
install/
最终安装结果:可执行文件、库、头文件、接口定义都会被安装到这里。
source install/setup.bash 后才能运行包里的节点。
log/
保存编译和运行时的日志,调试错误时常用。
Python常见类装饰器
1. @classmethod
@classmethod 装饰器用于将一个方法定义为类方法。类方法与普通实例方法不同,它接收的第一个参数是类本身,通常命名为 cls。这使得你可以在不创建类的实例的情况下调用这个方法。类方法通常用于创建工厂方法,或者执行与类相关的操作。
class MyClass:
count = 0
def __init__(self, name):
self.name = name
MyClass.count += 1
@classmethod
def get_count(cls):
return f"当前创建了 {cls.count} 个实例。"
obj1 = MyClass("A")
obj2 = MyClass("B")
print(MyClass.get_count())
# 输出: 当前创建了 2 个实例。
2. @staticmethod
@staticmethod 装饰器用于将一个方法定义为静态方法。静态方法既不接收类 (cls) 作为第一个参数,也不接收实例 (self) 作为第一个参数。它与类或实例的状态完全无关,更像是一个普通的函数,只是被组织在类的命名空间内。
class Calculator:
@staticmethod
def add(x, y):
return x + y
@staticmethod
def subtract(x, y):
return x - y
print(Calculator.add(5, 3))
# 输出: 8
3. @property
@property 装饰器将一个方法转换为只读属性。这使得你可以像访问属性一样调用方法,而无需使用括号。它通常用于将类的内部数据封装起来,提供一个更干净的接口。同时,你还可以配合 @<property_name>.setter 和 @<property_name>.deleter 来定义设置和删除属性的行为。
class Person:
def __init__(self, name):
self._name = name # 通常用下划线表示这是内部属性
@property
def name(self):
return self._name
@name.setter
def name(self, value):
if not isinstance(value, str):
raise TypeError("姓名必须是字符串。")
self._name = value
p = Person("Alice")
print(p.name) # 像访问属性一样调用
p.name = "Bob" # 使用setter修改
print(p.name)
4. @dataclass
@dataclass 装饰器是 Python 3.7 引入的,它主要用于快速创建数据类。数据类通常只包含数据,并且 dataclass 会自动为你生成许多有用的魔术方法,例如 __init__、__repr__、__eq__ 等,从而大大减少样板代码。
from dataclasses import dataclass
@dataclass
class Point:
x: float
y: float
p1 = Point(1.0, 2.0)
p2 = Point(1.0, 2.0)
print(p1) # 自动生成 __repr__
# 输出: Point(x=1.0, y=2.0)
print(p1 == p2) # 自动生成 __eq__
# 输出: True
5. @total_ordering
@total_ordering 装饰器位于 functools 模块中,它非常有用。如果你为一个类定义了至少一个富比较方法(__lt__, __le__, __gt__, __ge__),并且还定义了 __eq__,@total_ordering 就会自动为你生成其余的比较方法。这避免了重复编写所有比较逻辑。
from functools import total_ordering
@total_ordering
class Number:
def __init__(self, value):
self.value = value
def __eq__(self, other):
return self.value == other.value
def __lt__(self, other):
return self.value < other.value
n1 = Number(5)
n2 = Number(10)
print(n1 < n2) # True
print(n1 > n2) # True (由 @total_ordering 自动生成)
print(n1 <= n2) # True (由 @total_ordering 自动生成)
Python之多线程
多线程:在同一个进程中同时执行多个线程,提高程序的并发能力。
由于 Python 解释器的 GIL(全局解释器锁),多线程不能真正实现 CPU 并行计算,但对 I/O 密集型任务(网络请求、文件读写)非常有用。
GIL 是 CPython 解释器 为了保证线程安全而引入的一种机制,在任意时刻,只允许 一个线程 执行 Python 字节码,其本质:它是一个互斥锁,确保解释器内部的数据结构在多线程情况下不会出错。
Python之回调函数
回调函数:就是把一个函数作为参数传递给另一个函数,等前者任务执行完后再调用这个函数,常用于 异步编程、事件驱动,比如按钮点击、网络请求完成、ROS 订阅消息等。
import threading
t = threading.Thread(target=函数名, args=(参数1, 参数2, ...))
t.start() # 启动线程
t.join() # 等待线程执行完成(可选)
一般多线程和回调函数的配合使用
import threading
import requests
from bs4 import BeautifulSoup
class Download:
def download(self,url,callback_novel_name):
response = requests.get(url)
response.encoding = 'utf-8'
soup = BeautifulSoup(response.text, "html.parser")
title = soup.find("span", class_="txt").get_text()
score = soup.find("span", class_="score").get_text()
novel_name(title,score)
# 多线程下载小说
def strat(self,url,callback_novel_name):
thread = threading.Thread(target = self.download,args=(url,callback_novel_name))
thread.start()
# 回调函数
def novel_name(title,score):
print(f"title :{title},score :{score}")
if __name__=='__main__':
download = Download()
download.strat("https://www.qimao.com/shuku/1747899/",novel_name)
download.strat("https://www.qimao.com/shuku/1672986/",novel_name)
download.strat("https://www.qimao.com/shuku/1803136/",novel_name)

C++新特性
auto:auto 是 C++11 引入的一个非常实用的新特性,用来 自动类型推导。它让编译器根据初始化表达式自动推断变量的类型,省去了冗长的类型声明。
共享指针:std::shared_ptr 是一种智能指针,通过引用计数机制实现多个指针共享同一对象的所有权,并在最后一个指针销毁时自动释放内存,推荐 make_shared<T>()(效率高、异常安全)。
lambda表达式:C++中的 lambda表达式 是一种匿名函数,用于在需要时快速定义内联函数,常用于简化回调、算法和并发编程中的代码。
C++ lambda表达式 的一般格式是:
[capture](parameter_list) -> return_type {
function_body
};
/*
capture:捕获外部变量的方式,如值捕获 [=]、引用捕获 [&]、混合捕获 [a, &b]。
parameter_list:参数列表,类似普通函数参数。
return_type:返回类型,可省略,编译器会推导。
function_body:函数体,写逻辑代码。
*/
函数包装器:std::function 是 C++11 提供的通用函数包装器,用来以统一方式存储和调用各种可调用对象(函数、lambda、函数指针、仿函数等)。
std::function<返回类型(参数类型1, 参数类型2, ...)> 变量名;
#include <iostream>
#include <functional>
using namespace std;
int main(){
int a = 40;
int b = 20;
int c = 30;
function<void()> f;
f = [&]() -> void{
c = a + b;
};
auto ff = [=](int x,int y,int z) -> void{
cout<<x<<" "<<y<<" "<<z<<endl;
x = a + b;
y = a + c;
z = b + c;
cout<<x<<" "<<y<<" "<<z<<endl;
};
f();
cout<<a<<" "<<b<<" "<<c<<endl;
ff(a,b,c);
}
/*
lamda表达式的值有捕获和传参,因为虽然没有命名但是其本质也是一个函数不过lamda函数在原本函数传参的基础上有一个新的回去参数的方式那就是捕获
即会自动捕获其作用域范围内有效的lamba参数,其可捕获的作用域为定义lamda表达式确定的,定义之后的变量无法捕获,同时要注意引用捕获后面会影响前面的变量
*/
Lambda 捕获机制只与定义时的词法作用域有关,与调用顺序无关;但若捕获的是引用([&]),则调用时结果可能反映外部变量的最新状态。
C++的多线程使用标准库 <thread> 来创建线程,回调函数则使用函数包装器传递
C++的程序中一般叫主程序运行的线程为主线程,程序中通过thread创建的为子线程,二者为父子关系资源什么的都是共享的,子线程与父线程共享进程的全局资源(如全局变量、堆内存、文件描述符等),但拥有独立的栈空间和寄存器状态(因此局部变量不共享)。
子线程的不同启动方式:
注意点:
std::this_thread 是一个命名空间,提供当前线程的操作函数,即用于控制自身行为的函数
thread_local 是 C++11 引入的存储类型说明符,用于声明线程独有的变量:每个线程都有该变量的独立副本,互不干扰,避免共享资源的同步问题。
std::future 和 std::promise 用于获取异步任务的结果,避免直接操作线程。std::async 是创建异步任务的便捷函数,返回一个 std::future 对象,用于获取任务结果。
#include <iostream>
#include <thread>
#include <future>
#include <string>
#include <chrono>
#include <mutex> // 需包含互斥锁头文件
using namespace std;
using namespace chrono;
// 用引用传递string,并加锁保证线程安全
int complete_msg(string& msg, mutex& mtx) { // 注意:参数是引用
unique_lock<mutex> lock(mtx); // 加锁保护msg的访问
while(msg.empty()){
lock.unlock(); // 临时解锁,避免阻塞主线程修改msg
this_thread::sleep_for(seconds(1));
cout << "wait msg" << endl;
lock.lock(); // 重新加锁检查
}
return msg.length();
}
int main(){
future<int> result;
string msg;
mutex mtx; // 用于同步msg的互斥锁
// 传递msg的引用(用ref包装)和互斥锁
result = async(launch::async, complete_msg, ref(msg), ref(mtx));
this_thread::sleep_for(seconds(10));
// 主线程修改msg(需加锁)
{
lock_guard<mutex> lock(mtx); // 加锁保护修改
msg = "Hello, Async!";
}
// 获取结果(此时异步任务已退出循环)
cout << result.get() << endl; // 输出:14("Hello, Async!"的长度)
return 0;
}
#include <iostream>
#include <fstream>
#include <sstream>
#include <string>
#include <regex>
#include <thread>
#include <functional>
// 下载器类
class Download {
public:
void download(const std::string& filepath, std::function<void(const std::string&, const std::string&)> callback) {
// 读取本地文件
std::ifstream file(filepath);
if (!file.is_open()) {
std::cerr << "无法打开文件: " << filepath << std::endl;
return;
}
std::stringstream buffer;
buffer << file.rdbuf();
std::string content = buffer.str();
// 用正则匹配 title 和 score
std::regex title_regex(R"(<span\s+class="txt">(.*?)</span>)");
std::regex score_regex(R"(<span\s+class="score">(.*?)</span>)");
std::smatch match;
std::string title = "未知";
std::string score = "未知";
if (std::regex_search(content, match, title_regex)) {
title = match[1].str();
}
if (std::regex_search(content, match, score_regex)) {
score = match[1].str();
}
// 回调
callback(title, score);
}
void start(const std::string& filepath, std::function<void(const std::string&, const std::string&)> callback) {
std::thread t(&Download::download, this, filepath, callback);
t.detach(); // 分离线程,后台运行
}
};
// 回调函数
void novel_name(const std::string& title, const std::string& score) {
std::cout << "title: " << title << ", score: " << score << std::endl;
}
int main() {
Download downloader;
downloader.start("novel1.html", novel_name);
downloader.start("novel2.html", novel_name);
downloader.start("novel3.html", novel_name);
// 防止主线程提前退出
std::this_thread::sleep_for(std::chrono::seconds(2));
return 0;
}
总结
从项目的角度来看,一般来说一个可执行文件就是一个节点,作为ROS2系统中的最小执行单位,每个节点复杂不同的操作,像后续的话题,服务等都需要依靠节点来实现。


浙公网安备 33010602011771号