ROS2核心模塊

1.創建工作空間

先創建工作空間ws01_plumbing，

終端下進入工作空間的src目錄，執行如下命令：

ros2 pkg create --build-type ament_cmake base_interfaces_demo

2.話題通信

話題通信是ROS中使用頻率最高的一種通信模式，話題通信是基于發布訂閱模式的，也即：一個節點發布消息，另一個節點訂閱該消息。

話題通信是一種以發布訂閱的方式實現不同節點之間數據傳輸的通信模型。數據發布對象稱為發布方，數據訂閱對象稱之為訂閱方，發布方和訂閱方通過話題相關聯，發布方將消息發布在話題上，訂閱方則從該話題訂閱消息，消息的流向是單向的。

話題通信的發布方與訂閱方是一種多對多的關系，也即，同一話題下可以存在多個發布方，也可以存在多個訂閱方，這意味著數據會出現交叉傳輸的情況，當然如果沒有訂閱方，數據傳輸也會出現丟失的情況。

2.1話題通信案例

終端下進入工作空間的src目錄，調用如下兩條命令分別創建C++功能包和Python功能包。

ros2 pkg create cpp01_topic --build-type ament_cmake --dependencies rclcpp std_msgs base_interfaces_demo --node-name demo01_talker_str
ros2 pkg create py01_topic --build-type ament_python --dependencies rclpy std_msgs base_interfaces_demo --node-name demo01_talker_str_py

2.1.1 發送方案例（C++）

/*  需求：訂閱發布方發布的消息，并輸出到終端。步驟：1.包含頭文件；2.初始化 ROS2 客戶端；3.定義節點類；3-1.創建訂閱方；3-2.定時器。3-3.組織并發布消息4.調用spin函數，并傳入節點對象指針；5.釋放資源。
*/#include "rclcpp/rclcpp.hpp"
#include "std_msgs/msg/string.hpp"using namespace std::chrono_literals; //使用chrono庫中的時間單位//自定義節點類
class Talker:public rclcpp::Node{public:Talker():Node("Talker_Node_cpp"),count(0){//初始化count=0RCLCPP_INFO(this->get_logger(),"發布節點");// 3-1.創建訂閱方；publisher_=this->create_publisher<std_msgs::msg::String>("chatter",10);//創建一個名為"chater"的主題，隊列長度為10// 3-2.定時器。// 參數：時間間隔，毀掉函數，返回值：當前對象指針timer_=this->create_wall_timer(1s,std::bind(&Talker::on_timer,this));//創建一個定時器，每隔1秒調用一次on_timer函數}private:size_t count;//定義一個計數器，用于記錄發布的消息數量void on_timer(){// 3-3.組織并發布消息auto msg=std_msgs::msg::String();//創建一個String類型的消息對象msg.data="Hello ROS2 C++! "+std::to_string(count++);//給消息對象賦值RCLCPP_INFO(this->get_logger(),"發布消息：%s",msg.data.c_str());//輸出日志信息publisher_->publish(msg);}rclcpp::Publisher<std_msgs::msg::String>::SharedPtr publisher_;//定義一個發布者指針，類型為std_msgs::msg::Stringrclcpp::TimerBase::SharedPtr timer_;//定義一個定時器指針
};int main(int argc,char** argv){//初始化rclcpp::init(argc,argv);//調用spin,掛起當前rclcpp::spin(std::make_shared<Talker>());//釋放rclcpp::shutdown();
}

在終端運行

colcon build --packages-select cpp01_topic
. install/setup.bash
ros2 run cpp01_topic demo01_talker_str

可以另外一個終端去驗證chatter主題下是否有內容

. install/setup.bash
ros2 topic echo /chatter

2.1.2 訂閱方案例（C++）

/*  需求：訂閱發布方發布的消息，并輸出到終端。步驟：1.包含頭文件；2.初始化 ROS2 客戶端；3.定義節點類；3-1.創建訂閱方；3-2.定時器。3-3.組織并發布消息4.調用spin函數，并傳入節點對象指針；5.釋放資源。
*/#include "rclcpp/rclcpp.hpp"
#include "std_msgs/msg/string.hpp"using namespace std::chrono_literals; //使用chrono庫中的時間單位//自定義節點類
class Talker:public rclcpp::Node{public:Talker():Node("Talker_Node_cpp"),count(0){//初始化count=0RCLCPP_INFO(this->get_logger(),"發布節點");// 3-1.創建訂閱方；publisher_=this->create_publisher<std_msgs::msg::String>("chatter",10);//創建一個名為"chater"的主題，隊列長度為10// 3-2.定時器。// 參數：時間間隔，毀掉函數，返回值：當前對象指針timer_=this->create_wall_timer(1s,std::bind(&Talker::on_timer,this));//創建一個定時器，每隔1秒調用一次on_timer函數}private:size_t count;//定義一個計數器，用于記錄發布的消息數量void on_timer(){// 3-3.組織并發布消息auto msg=std_msgs::msg::String();//創建一個String類型的消息對象msg.data="Hello ROS2 C++! "+std::to_string(count++);//給消息對象賦值RCLCPP_INFO(this->get_logger(),"發布消息：%s",msg.data.c_str());//輸出日志信息publisher_->publish(msg);}rclcpp::Publisher<std_msgs::msg::String>::SharedPtr publisher_;//定義一個發布者指針，類型為std_msgs::msg::Stringrclcpp::TimerBase::SharedPtr timer_;//定義一個定時器指針
};int main(int argc,char** argv){//初始化rclcpp::init(argc,argv);//調用spin,掛起當前rclcpp::spin(std::make_shared<Talker>());//釋放rclcpp::shutdown();
}

需要去CMakeLists.txt三處添加demo02_listener_str

在終端運行（要先啟動發送方，訂閱方才有信息）

colcon build --packages-select cpp01_topic
. install/setup.bash
ros2 run cpp01_topic demo02_listener_str

2.1.3 發送方案例（python）

     # 需求：訂閱發布方發布的消息，并輸出到終端。# 步驟：#     1.包含頭文件；#     2.初始化 ROS2 客戶端；#     3.定義節點類；#         3-1.創建訂閱方；#         3-2.定時器。#         3-3.組織并發布消息#     4.調用spin函數，并傳入節點對象指針；#     5.釋放資源。import rclpy
from rclpy.node import Node
from std_msgs.msg import String# 定義節點類
class Puublisher(Node):def __init__(self):super().__init__('talker_node_py')self.get_logger().info('發布節點')# 3-1.創建發布方self.publisher_ = self.create_publisher(String, 'chatter', 10)# 3-2.定時器self.timer = self.create_timer(1.0, self.on_timer)self.count = 0def on_timer(self):# 3-3.組織并發布消息msg = String()msg.data = f"Hello ROS2 Python! {self.count}"self.get_logger().info(f'發布消息：{msg.data}')self.publisher_.publish(msg)self.count += 1def main():# 初始化 ROS2 客戶端rclpy.init()# 創建節點對象node = Puublisher()# 調用 spin 函數，傳入節點對象指針rclpy.spin(node)# 釋放資源rclpy.shutdown()if __name__ == '__main__':main()

2.1.4?訂閱方案例（python）

    # 需求：訂閱發布方發布的消息，并在終端輸出# 流程：#     1.頭文件#     2.初始化客戶端#     3.自定義節點類#         3-1.創建訂閱方#         3-2.解析并輸出數據#     4.調用spin函數，傳入節點對象指針#     5.釋放資源import rclpy
from rclpy.node import Node
from std_msgs.msg import String# 自定義節點類
class Subscriber(Node):def __init__(self):super().__init__('listener_node_py')# 創建訂閱方self.subscription=self.create_subscription(String,"chatter",self.listen_callback,10,)self.subscription  # 訂閱對象指針#處理訂閱到的消息def listen_callback(self, msg):# 解析并輸出數據self.get_logger().info(f'訂閱到消息：{msg.data}')   def main(args=None):# 初始化 ROS2 客戶端rclpy.init(args=args)# 創建節點對象node = Subscriber()# 調用 spin 函數，傳入節點對象指針rclpy.spin(node)# 釋放資源rclpy.shutdown()if __name__ == '__main__':main()

訂閱方創建好之后需要到setup.py文件下修改entry_points參數，將demo02加進去：

然后編譯運行

colcon build --packages-select py01_topic
. install/setup.bash
ros2 run py01_topic demo01_talker_str_py//再重開一個終端運行
. install/setup.bash
ros2 run py01_topic demo02_listener_str_py

2.1.5?自定義接口消息

創建并編輯 .msg 文件

功能包base_interfaces_demo下新建 msg 文件夾，msg文件夾下新建Student.msg文件，文件中輸入如下內容：

string   name
int32    age
float64  height

在package.xml文件添加

如果有報錯的直接刪掉

在CMakeList.txt添加

在終端編譯測試

colcon build --packages-select base_interfaces_demo. install/setup.bash
ros2 interface show base_interfaces_demo/msg/Student

輸出內容和Student.msg一致

2.1.6 自定義發布方（C++）

/*需求：以固定頻率發布學生信息。步驟：1.包含頭文件；2.初始化 ROS2 客戶端；3.定義節點類；3-1創建發布方3-2創建定時器3-3組織并發布消息4.調用spin函數，并傳入節點對象指針；5.釋放資源。
*/
#include <rclcpp/rclcpp.hpp>
#include "std_msgs/msg/string.hpp"
#include "base_interfaces_demo/msg/student.hpp" // 引入自定義消息類型using base_interfaces_demo::msg::Student; // 使用自定義消息類型的命名空間
using namespace std::chrono_literals; // 使用 chrono 的時間單位
// 自定義節點類
class TalkStu : public rclcpp::Node
{
public:TalkStu() : Node("TalkStu_node_cpp"){// 3-1創建發布方publisher_=this->create_publisher<Student>("chat_stu", 10); // 創建一個名為 "chat_stu" 的發布者，隊列長度為 10// 3-2創建定時器timer_=this->create_wall_timer(1s,std::bind(&TalkStu::timer_callback, this)); // 每秒調用一次 timer_callback 函數                }private:int age=0;void timer_callback(){// 3-3組織并發布消息auto stu=Student();stu.name="張三";stu.age=age++;stu.height=1.75;age++;publisher_->publish(stu); // 發布消息RCLCPP_INFO(this->get_logger(), "發布學生信息: name=%s,age=%d,height=%.2f)",stu.name.c_str(), stu.age, stu.height);};rclcpp::Publisher<Student>::SharedPtr publisher_; // 定義發布者指針rclcpp::TimerBase::SharedPtr timer_; // 定義定時器指針};int main(int argc, char const *argv[])
{// 初始化rclcpp::init(argc, argv);// 調用spin,掛起當前rclcpp::spin(std::make_shared<TalkStu>());// 釋放rclcpp::shutdown();return 0;
}

package.xml無需更改，需要在CMakeList.txt文件下添加這兩個文件

2.1.7 自定義訂閱方（C++）

/*  需求：訂閱發布方發布的消息，并輸出到終端。步驟：1.包含頭文件；2.初始化 ROS2 客戶端；3.定義節點類；3-1創建訂閱方；3-2創建回調函數；訂閱并解析數據4.調用spin函數，并傳入節點對象指針；5.釋放資源。
*/
#include <rclcpp/rclcpp.hpp>
#include "std_msgs/msg/string.hpp"
#include "base_interfaces_demo/msg/student.hpp"using base_interfaces_demo::msg::Student; // 使用自定義消息類型的命名空間//自定義節點類
class ListenerStu:public rclcpp::Node{
public:ListenerStu():Node("ListenerStu_node_cpp"){// 3-1創建訂閱方；subscription_=this->create_subscription<Student>("chat_stu",10,std::bind(&ListenerStu::topic_callback,this,std::placeholders::_1));}
private:void topic_callback(const Student &stu){// 3-2創建回調函數；訂閱并解析數據RCLCPP_INFO(this->get_logger(),"訂閱信息: name=%s, age=%d, height=%.2f",stu.name.c_str(), stu.age, stu.height);};rclcpp::Subscription<Student>::SharedPtr subscription_; // 定義訂閱者指針
};int main(int argc,char const *argv[]){//初始化rclcpp::init(argc,argv);//調用spin,掛起當前rclcpp::spin(std::make_shared<ListenerStu>());//釋放rclcpp::shutdown();
}

在兩個終端分別運行

colcon build --packages-select cpp01_topic
. install/setup.bash
ros2 run cpp01_topic demo03_talker_stu. install/setup.bash
ros2 run cpp01_topic demo04_listener_stu

2.1.8 自定義發布方（python）

"""
流程：1. 導包2. 初始化ROS2客戶端3. 自定義節點類3-1. 創建發布者/訂閱者3-2. 處理數據（回調函數）4. 調用 spin 函數，傳入節點對象5. 資源釋放
"""import rclpy
from rclpy.node import Node
from base_interfaces_demo.msg import Student # 可根據需要替換為其他消息類型class TalkerStu(Node):def __init__(self):super().__init__('my_node_py')self.get_logger().info("節點已啟動:my_node_py")# 3-1. 創建發布者（可選）self.publisher_ = self.create_publisher(Student, 'chat_stu', 10)self.timer = self.create_timer(0.5, self.timer_callback)  # 每0.5秒發布一次self.count=0# 3-2. 處理數據def timer_callback(self):msg = Student()msg.name = '張三'msg.age = self.countmsg.height=1.72self.publisher_.publish(msg)self.count+=1self.get_logger().info('發布消息: name=%s,age=%d,height=%.2f' % (msg.name, msg.age, msg.height))def main(args=None):# 2. 初始化ROS2客戶端rclpy.init(args=args)# 3. 創建節點對象node = TalkerStu()# 4. 調用 spin，傳入節點對象rclpy.spin(node)# 5. 釋放資源rclpy.shutdown()if __name__ == '__main__':main()

from base_interfaces_demo.msg import Student # 可根據需要替換為其他消息類型

這一段需要crtl+shift+p搜索dettings.json
然后在文件種添加

{"python.autoComplete.extraPaths": ["/home/sun/ws01_pluming/install/base_interfaces_demo/local/lib/python3.10/dist-packages",],"python.analysis.extraPaths": ["/home/sun/ws01_pluming/install/base_interfaces_demo/local/lib/python3.10/dist-packages",]
}

然后Student才會生效。

之后進入到setup.py文件下添加

2.1.9 自定義訂閱放（python）

"""
流程：1. 導包2. 初始化ROS2客戶端3. 自定義節點類3-1. 創建發布者/訂閱者3-2. 處理數據（回調函數）4. 調用 spin 函數，傳入節點對象5. 資源釋放
"""import rclpy
from rclpy.node import Node
from base_interfaces_demo.msg import Student   # 可根據需要替換為其他消息類型class ListenerStu(Node):def __init__(self):super().__init__('my_node_py')self.get_logger().info('節點已啟動：my_node_py')# 3-1. 創建訂閱者（可選）self.subscription = self.create_subscription(Student,'chat_stu',self.listener_callback,10)self.subscription  # 防止未使用警告# 3-2. 處理數據def listener_callback(self, msg):self.get_logger().info('收到消息: name=%s,age=%d,height=%.2f' % (msg.name, msg.age, msg.height))def main(args=None):# 2. 初始化ROS2客戶端rclpy.init(args=args)# 3. 創建節點對象node = ListenerStu()# 4. 調用 spin，傳入節點對象rclpy.spin(node)# 5. 釋放資源rclpy.shutdown()if __name__ == '__main__':main()

然后在終端運行

colcon build --packages-select py01_topic
. install/setup.bash
ros2 run py03_topic demo03_talker_stu_py. install/setup.bash
ros2 run py04_topic demo04_listener_stu_py