这里写目录标题

1.通信机制简介

在这里插入图片描述

在ROS2中通信方式虽然有多种，但是不同通信方式的组成要素都是类似的，比如：通信是双方或多方行为、通信时都需要将不同的通信对象关联、都有各自的模型、交互数据时也必然涉及到数据载体等等。本节将会介绍通信中涉及到的一些术语。

1.1.节点

在通信时，不论采用何种方式，通信对象的构建都依赖于节点(Node)，在ROS2中，一般情况下每个节点都对应某一单一的功能模块(例如：雷达驱动节点可能负责发布雷达消息，摄像头驱动节点可能负责发布图像消息)。一个完整的机器人系统可能由许多协同工作的节点组成，ROS2中的单个可执行文件(C++程序或Python程序)可以包含一个或多个节点。

1.2.话题

话题(Topic)是一个纽带，具有相同话题的节点可以关联在一起，而这正是通信的前提。并且ROS2是跨语言的，有的节点可能是使用C++实现，有的节点可能是使用Python实现的，但是只要二者使用了相同的话题，就可以实现数据的交互。

1.3.通信模型

在这里插入图片描述

不同的通信对象通过话题关联到一起之后，以何种方式实现通信呢？在ROS2中，常用的通信模型有四种：

1.话题通信：是一种单向通信模型，在通信双方中，发布方发布数据，订阅方订阅数据，数据流单向的由发布方传输到订阅方。

2.服务通信：是一种基于请求响应的通信模型，在通信双方中，客户端发送请求数据到服务端，服务端响应结果给客户端。

3.动作通信：是一种带有连续反馈的通信模型，在通信双方中，客户端发送请求数据到服务端，服务端响应结果给客户端，但是在服务端接收到请求到产生最终响应的过程中，会发送连续的反馈信息到客户端。

4.参数服务：是一种基于共享的通信模型，在通信双方中，服务端可以设置数据，而客户端可以连接服务端并操作服务端数据。

1.4.接口

在通信过程中，需要传输数据，就必然涉及到数据载体，也即要以特定格式传输数据。在ROS2中，数据载体称之为接口(interfaces)。通信时使用的数据载体一般需要使用接口文件定义。常用的接口文件有三种：msg文件、srv文件与action文件。每种文件都可以按照一定格式定义特定数据类型的“变量”。

  • 1.msg文件

msg文件是用于定义话题通信中数据载体的接口文件，一个典型的文件示例如下。

int64 num1
int64 num2

在文件中声明了一些被传输的类似于C++变量的数据。

  • 2.srv文件

srv文件是用于定义服务通信中数据载体的接口文件，一个典型的文件示例如下。

int64 num1
int64 num2
---
int64 sum

文件中声明的数据被分割为两部分，上半部分用于声明请求数据，下半部分用于声明响应数据。

  • 3.action文件

action文件使用用于定义动作通信中数据载体的接口文件，一个典型的文件示例如下。

int64 num1
---
int64 num2
---
float progress

文件中声明的数据被分割为三部分，上半部分用于声明请求数据，中间部分用于声明响应数据，下半部分用于声明连续反馈数据。

  • 4.变量类型

不管是何种接口文件，在文件中每行声明的数据都由字段类型和字段名称组成，可以使用的字段类型有：

    int8, int16, int32, int64 (或者无符号类型: uint*)
    float32, float64
    string
    time, duration
    其他msg文件
    变长数组和定长数组

ROS中还有一种特殊类型：Header，标头包含时间戳和ROS2中常用的坐标帧信息。许多接口文件的第一行包含标头。

另外，需要说明的是：

参数通信的数据无需定义接口文件，参数通信时数据会被封装为参数对象，参数客户端和服务端操作的都是参数对象。

2 实例应用

2.0 准备工作

在这里插入图片描述

2.１ 话题通讯

2.1.1场景

话题通信是ROS中使用频率最高的一种通信模式，话题通信是基于发布订阅模式的，也即：一个节点发布消息，另一个节点订阅该消息。话题通信的应用场景也极其广泛，比如如下场景：

机器人在执行导航功能，使用的传感器是激光雷达，机器人会采集激光雷达感知到的信息并计算，
然后生成运动控制信息驱动机器人底盘运动。

在该场景中，就不止一次使用到了话题通信。

  • 以激光雷达信息的采集处理为例，在ROS中有一个节点需要时时的发布当前雷达采集到的数据，导航模块中也有节点会订阅并解析雷达数据。
  • 再以运动消息的发布为例，导航模块会综合多方面数据实时计算出运动控制信息并发布给底盘驱动模块，底盘驱动有一个节点订阅运动信息并将其转换成控制电机的脉冲信号。
    以此类推，像雷达、摄像头、GPS…等等一些传感器数据的采集 ，也都是使用了话题通信，话题通信适用于不断更新的数据传输相关的应用场景。

2.1.2 概念

话题通信是一种以发布订阅的方式实现不同节点之间数据传输的通信模型。数据发布对象称为发布方，数据订阅对象称之为订阅方，发布方和订阅方通过话题相关联，发布方将消息发布在话题上，订阅方则从该话题订阅消息，消息的流向是单向的。

在这里插入图片描述

话题通信的发布方与订阅方是一种多对多的关系，也即，同一话题下可以存在多个发布方，也可以存在多个订阅方，这意味着数据会出现交叉传输的情况，当然如果没有订阅方，数据传输也会出现丢失的情况。

在这里插入图片描述

2.1.3 作用

话题通信一般应用于不断更新的、少逻辑处理的数据传输场景。

  • 关于消息接口

关于消息接口的使用有多种方式：

  • 在ROS2中通过std_msgs包封装了一些原生的数据类型,比如：String、Int8、Int16、Int32、Int64、Float32、Float64、Char、Bool、Empty.... 这些原生数据类型也可以作为话题通信的载体，不过这些数据一般只包含一个 data 字段，而std_msgs包中其他的接口文件也比较简单，结构的单一意味着功能上的局限性，当传输一些结构复杂的数据时，就显得力不从心了；
  • 在ROS2中还预定义了许多标准话题消息接口，这在实际工作中有着广泛的应用，比如：sensor_msgs包中定义了许多关于传感器消息的接口（雷达、摄像头、点云......），geometry_msgs包中则定义了许多几何消息相关的接口（坐标点、坐标系、速度指令......）；
  • 如果上述接口文件都不能满足我们的需求，那么就可以自定义接口消息；
    具体如何选型，大家可以根据具体情况具体分析。

2.1.4 案例

2.1.4.1 案例分析
1.案例需求

需求1：编写话题通信实现，发布方以某个频率发布一段文本，订阅方订阅消息，并输出在终端。

需求2：编写话题通信实现，发布方以某个频率发布自定义接口消息，订阅方订阅消息，并输出在终端。

2.案例分析

在上述案例中，需要关注的要素有三个：

发布方；
订阅方；
消息载体。

案例1和案例2的主要区别在于消息载体，前者可以使用原生的数据类型，后者需要自定义接口消息。

3.流程简介

案例2需要先自定义接口消息，除此之外的实现流程与案例1一致，主要步骤如下：

s1:编写发布方实现；
s2:编写订阅方实现；
s3:编辑配置文件；
s4:编译；
s5:执行。

案例我们会主要采用Python分别实现，遵循上述实现流程。

2.1.4.2 话题通信之原生消息（Python）
0.准备工作:创建功能包

终端下进入工作空间的src目录，调用如下两条命令创建Python功能包。

ros2 pkg create py01_topic --build-type ament_python --dependencies rclpy std_msgs base_interfaces_demo --node-name demo01_talker_str_py

注意：上面依赖了三个功能包

1.发布方实现

在这里插入图片描述

功能包py01_topic的py01_topic目录下，新建Python文件demo01_talker_str_py.py，并编辑文件，输入如下内容：

#1.导入使用工具包
import rclpy
from rclpy.node import Node
from std_msgs.msg import String

# 3.定义节点类；
class MinimalPublisher(Node):
    def __init__(self):
        super().__init__('minimal_publisher_py')
        #3-1.创建一个发布者，主题为'demo01_chatter_str'，消息类型为String,10表示队列长度
        self.publisher_ = self.create_publisher(String, 'topic', 10)
        #3-2.创建一个定时器，每0.5秒调用一次回调函数
        timer_period = 0.5  # 发布周期为0.5秒
        self.timer = self.create_timer(timer_period, self.timer_callback)
        self.count = 0

    def timer_callback(self):
        # 3-3.定义一个计数器，用于发布消息内容
        msg = String()
        msg.data = f'Hello World: {self.count}'
        self.publisher_.publish(msg)
        self.get_logger().info(f'Publishing: "{msg.data}"')
        self.count += 1

def main(args=None):
    # 2.初始化ROS2 Python客户端库
    rclpy.init(args=args)

    # 4.创建节点实例
    minimal_publisher = MinimalPublisher()

    # 5.使用rclpy.spin()方法进入循环，等待回调函数被调用
    rclpy.spin(minimal_publisher)

    # 6.销毁节点
    minimal_publisher.destroy_node()
    
    # 7.关闭ROS2 Python客户端库
    rclpy.shutdown()
if __name__ == '__main__':
    main()
2.订阅方实现

功能包py01_topic的py01_topic目录下，新建Python文件demo02_listener_str_py.py，并编辑文件，输入如下内容：

在这里插入图片描述

3.编辑配置文件

在Python功能包中，配置文件主要关注package.xml与setup.py。

  • 1.package.xml
    在创建功能包时，所依赖的功能包已经自动配置了，配置内容如下：
<?xml version="1.0"?>
<?xml-model href="http://download.ros.org/schema/package_format3.xsd" schematypens="http://www.w3.org/2001/XMLSchema"?>
<package format="3">
  <name>py01_topic</name>
  <version>0.0.0</version>
  <description>TODO: Package description</description>
  <maintainer email="ztfmars@163.com">fusionai</maintainer>
  <license>TODO: License declaration</license>

  <depend>rclpy</depend>
  <depend>std_msgs</depend>
  <depend>base_interfaces_demo</depend>

  <test_depend>ament_copyright</test_depend>
  <test_depend>ament_flake8</test_depend>
  <test_depend>ament_pep257</test_depend>
  <test_depend>python3-pytest</test_depend>

  <export>
    <build_type>ament_python</build_type>
  </export>
</package>
  • 2.setup.py
    字段的中添加如下内容：
    'demo02_listener_str_py = py01_topic.demo02_listener_str_py:main'
    完整如下:
from setuptools import find_packages, setup

package_name = 'py01_topic'

setup(
    name=package_name,
    version='0.0.0',
    packages=find_packages(exclude=['test']),
    data_files=[
        ('share/ament_index/resource_index/packages',
            ['resource/' + package_name]),
        ('share/' + package_name, ['package.xml']),
    ],
    install_requires=['setuptools'],
    zip_safe=True,
    maintainer='fusionai',
    maintainer_email='ztfmars@163.com',
    description='TODO: Package description',
    license='TODO: License declaration',
    tests_require=['pytest'],
    entry_points={
        'console_scripts': [
            'demo01_talker_str_py = py01_topic.demo01_talker_str_py:main',
            'demo02_listener_str_py = py01_topic.demo02_listener_str_py:main'
        ],
    },
)
4.编译

终端中进入当前工作空间，编译功能包：

colcon build --packages-select py01_topic
5.执行

当前工作空间下，启动两个终端，终端1执行发布程序，终端2执行订阅程序。

# 激活环境
source ./install/setup.bash

终端1输入如下指令：

ros2 run py01_topic demo01_talker_str_py

在这里插入图片描述
也可使用ros2中topic进行内容输出:

# /topic表示正在打印的topic内容，名称跟self.publisher_ = self.create_publisher(String, 'topic', 10)中'topic'内容一致
#　发布和接受者可以跨语言
ros2 topic echo /topic

在这里插入图片描述

终端2输入如下指令：

ros2 run py01_topic demo02_listener_str_py

最终运行结果与案例1类似。
在这里插入图片描述

2.1.4.3 话题通信之自定义接口消息（Python）

自定义接口消息的流程与在功能包中编写可执行程序的流程类似，主要步骤如下：

<font color=red>（cpp创建）</font>创建并编辑 .msg文件；
编辑配置文件；
编译；
测试。

接下来，我们可以参考案例2编译一个msg文件，该文件中包含学生的姓名、年龄、身高等字段。

1.准备

Python文件中导入自定义消息相关的包时，为了方便使用，可以配置VSCode中settings.json文件，在文件中的python.autoComplete.extraPaths和python.analysis.extraPaths属性下添加一行：&#34;${workspaceFolder}/install/base_interfaces_demo/local/lib/python3.10/dist-packages&#34;
添加完毕后，代码可以高亮显示且可以自动补齐，其他接口文件或接口包的使用也与此同理。

2.自定义消息的创建
  • 1 ) 先创建ws01_plumbing,本章以及第3部分代码内容都在该工作空间下．
  • ２) 实际应用中一般建议创建专用接口功能包定义接口文件，当前教程也遵循这一建议，预先创建教程所需使用接口功能包(！！！需要注意的是到目前位置，无法在python功能包中定义接口文件，需要使用cpp先创建相关接口文件（base_interfaces_demo），编译之后再供python使用！！！)．

a.创建自定义消息功能包
终端下进入工作空间的src目录，执行如下命令：

ros2 pkg create --build-type ament_cmake base_interfaces_demo

该功能包将用于保存本章教程中自定义的接口文件．

相关cpp的内容，参见　CPP版本实现

b.创建并编辑 .msg 文件
功能包base_interfaces_demo下新建 msg 文件夹，msg文件夹下新建Student.msg文件，文件中输入如下内容：

string name
int32 age
float64 height

个人理解：相当于是ros2消息版的 c/c++结构体 的抽象实现

  • 3 ). package.xml 文件

在package.xml中需要添加一些依赖包，具体内容如下：

#####  <depend>action_msgs</depend>
##### .... 在此处
#####   <export>
####    <build_type>ament_cmake</build_type>

  <!-- 编译依赖 -->
  <build_depend>rosidl_default_generators</build_depend>

  <test_depend>ament_lint_auto</test_depend>
  <test_depend>ament_lint_common</test_depend>

  <!-- 执行依赖 -->
  <exec_depend>rosidl_default_runtime</exec_depend>

  <!-- 声明当前包所属的功能包组 -->
  <member_of_group>rosidl_interface_packages</member_of_group>
  • 4 ). CMakeLists.txt文件
    为了将.msg文件转换成对应的C++和Python代码，还需要在CMakeLists.txt中添加如下配置：
###### find dependencies
###### find_package(ament_cmake REQUIRED) (在此之后添加)

find_package(rosidl_default_generators REQUIRED)
 
rosidl_generate_interfaces(${PROJECT_NAME}
  "msg/Student.msg"
)
  • 5 ) 终端中进入当前工作空间，编译功能包：
colcon build --packages-select base_interfaces_demo
  • 6 ) 测试

编译完成之后，在工作空间下的install目录下将生成Student.msg文件对应的C++和Python文件。

1
2

我们也可以在终端下进入工作空间，通过如下命令查看文件定义以及编译是否正常：

. install/setup.bash
ros2 interface show base_interfaces_demo/msg/Student

正常情况下，终端将会输出与Student.msg文件一致的内容。
3

总结：接口文件 msg c/c+ 中的结构体 自定义消息

3. 自定义消息的发布和订阅
  • 发布＆订阅方实现
    在这里插入图片描述

a.发布方实现
功能包py01_topic的py01_topic目录下，新建Python文件demo03_talker_stu_py.py，并编辑文件，输入如下内容：

"""
需求：以固定频率发布学生信息
步骤:
    1.导包
    2.初始化 ROS2 客户端
    3.定义节点类
        3-1.创建消息发布方
        3-2.创建定时器
        3-3.组织消息并发布学生信息
    4.调用spin函数,并传入节点对象
    5.释放资源
"""

import rclpy
from rclpy.node import Node
from base_interfaces_demo.msg import Student

# 3.定义节点类
class TalkerStu(Node):
    def __init__(self):
        super().__init__("talker_stu_node_py")
        self.get_logger().info("发布方创建了...")

        # 3-1.创建消息发布方
        self.publisher = self.create_publisher(Student, "chatter_stu", 10)

        # 3-2.创建定时器
        self.timer = self.create_timer(0.5, self.on_timer)

    def on_timer(self):
        # 3-3.组织消息并发布
        stu = Student()
        stu.name = "李四"
        stu.age = 21
        stu.height = 1.7
        self.publisher.publish(stu)
        self.get_logger().info(f"发布的数据:{stu}")


def main(args=None):
    rclpy.init(args=args)
    rclpy.spin(TalkerStu())
    rclpy.shutdown()


if __name__ == '__main__':
    main()

b.订阅方实现
功能包py01_topic的py01_topic目录下，新建Python文件demo04_listener_stu_py.py，并编辑文件，输入如下内容：

"""
需求：订阅发布方发布的消息，并在终端输出
流程：
    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

# 3.自定义节点类
class ListenerStu(Node):
    def __init__(self):
        super().__init__("listener_stu_node_py")
        self.get_logger().info("订阅方创建了...")

        # 3-1.创建订阅方
        self.subscription = self.create_subscription(
            Student, "chatter_stu", self.do_cb, 10
        )

    def do_cb(self, stu):
        """
        回调函数 \n
        :param stu: 接收到的学生消息
        """

        # 3-2.解析并输出数据
        self.get_logger().info(f"name={stu.name} age={stu.age} height={stu.height}")


def main():
    rclpy.init()
    rclpy.spin(ListenerStu())
    rclpy.shutdown()


if __name__ == "__main__":
    main()

  • 编辑配置文件
    package.xml无需修改，需要修改setup.py文件，字段的中修改为如下内容：
    在这里插入图片描述

  • 编译
    终端中进入当前工作空间，编译功能包：

colcon build --packages-select py01_topic
  • 执行
    当前工作空间下，启动两个终端，终端1执行发布程序，终端2执行订阅程序。

终端1输入如下指令：

source ./intstall/setup.bash
ros2 run py01_topic demo03_talker_stu_py

在这里插入图片描述

终端2输入如下指令：

source ./intstall/setup.bash
ros2 run py01_topic demo04_listener_stu_py

最终运行结果与案例2类似。

在这里插入图片描述

2.1.4.４ 图形化工具－ＲＱＴ

首先参考之前的内容，启动一个（或几个）发布topic的talker,　启动另一个（或几个）接受topic的listener．查看相关节点信息和关联．

### 第一个窗口talker
source ./intstall/setup.bash
ros2 run py01_topic demo03_talker_stu_py


### 第二个窗口listener
source ./intstall/setup.bash
ros2 run py01_topic demo04_listener_stu_py

###　第三个窗口rqt
rqt

在这里插入图片描述

之后在显示的rqt界面中，查找相关Ｎｏｄｅ内容
在这里插入图片描述

可以看到对应的单点Node节点
在这里插入图片描述

2.2 服务通讯

2.2.1 场景

服务通信也是ROS中一种极其常用的通信模式，服务通信是基于请求响应模式的，是一种应答机制。也即：一个节点A向另一个节点B发送请求，B接收处理请求并产生响应结果返回给A。比如如下场景：

机器人巡逻过程中，控制系统分析传感器数据发现可疑物体或人... 此时需要拍摄照片并留存。

在上述场景中，就使用到了服务通信。

数据分析节点A需要向相机相关节点B发送图片存储请求，节点B处理请求，并返回处理结果。
与上述应用类似的，服务通信更适用于对实时性有要求、具有一定逻辑处理的应用场景。

2.2.2 概念

服务通信是以请求响应的方式实现不同节点之间数据传输的通信模式。发送请求数据的对象称为客户端，接收请求并发送响应的对象称之为服务端，同话题通信一样，客户端和服务端也通过话题相关联，不同的是服务通信的数据传输是双向交互式的。

xxx

服务通信中，服务端与客户端是一对多的关系，也即，同一服务话题下，存在多个客户端，每个客户端都可以向服务端发送请求。

xx

2.2.3 作用

用于偶然的、对实时性有要求、有一定逻辑处理需求的数据传输场景。

2.2.4 案例以及案例分析

1.案例需求
需求：编写服务通信，客户端可以提交两个整数到服务端，服务端接收请求并解析两个整数求和，然后将结果响应回客户端。

xxx

2.案例分析
在上述案例中，需要关注的要素有三个：

客户端；
服务端；
消息载体。

3.流程简介
案例实现前需要先自定义服务接口，接口准备完毕后，服务实现主要步骤如下：

编写服务端实现；
编写客户端实现；
编辑配置文件；
编译；
执行。

案例我们会采用Python分别实现，二者都遵循上述实现流程。

因为相关消息接口只能使用cpp，所以我们需要先定义相关消息接口／编译之后，然后在编译相关python消息发布方和接受方．

步骤１：创建功能包

我们直接在上面2.1.4自定义话题文件夹ws01_plumbing/src文件夹下创建新的服务通信内容:

# 注意：相关以来有rclpy & base_interfaces_demo
ros2 pkg create py02_service --build-type ament_python --dependencies rclpy base_interfaces_demo --node-name demo01_server_py

在这里插入图片描述

步骤２：创建服务通信接口消息＆编辑相关文件

在这里插入图片描述

  • 创建编辑srv文件　＆　修改配置文件

在这里插入图片描述

在这里插入图片描述

  • 编译
    终端进入工作空间，编译功能包:
colcon build --packages-select base_interfaces_demo

编译完成之后，在/ws01_plumbing/install/base_interfaces_demo/include/base_interfaces_demo/base_interfaces_demo/srv目录下生成相关接口信息的文件

  • 测试
    在这里插入图片描述
步骤３：服务通信(python)
  • 新建一个py01_server工作包
ros2 pkg create py02_service --build-type ament_python --dependencies rclpy base_interfaces_demo --node-name demo01_server
  • 添加发送／接收端代码
#### demo01_server.py

"""
需求:创建服务端,解析客户端提交的数据并响应
流程:
    3.自定义节点
        3-1.创建服务类
        3-2.编写回调函数,处理请求并产生响应
    4.调用spin函数,并传入节点对象
"""

import rclpy
from rclpy.node import Node
from base_interfaces_demo.srv import AddInts


class AddIntsServer(Node):
    def __init__(self):
        super().__init__("add_ints_server_node_py")
        self.get_logger().info("服务端已创建")
        # 创建服务端
        self.create_service(AddInts, "add_ints", self.add)

    def add(self, request, response):
        response.sum = request.num1 + request.num2
        self.get_logger().info(f"{request.num1} + {request.num2} = {response.sum}")
        return response


def main():
    rclpy.init()
    rclpy.spin(AddIntsServer())
    rclpy.shutdown()


if __name__ == '__main__':
    main()

#### demo01_client.py
"""
需求:编写客户端,发送两个整型变量作为请求数据,并处理响应结果
流程:
    1.前提:main函数中判断提交的参数是否正确
    2.初始化 ROS2 客户端
    3.自定义节点类:
        3-1.创建客户端
        3-2.连接服务器(如果客户端无法连接到服务端,则不能发送请求)
        3-3.发送请求
    4.创建对象
        4-1.调用连接服务的函数,根据连接结果做进一步处理
        4-2.连接服务后,调用请求发送函数
        4-3.再处理响应结果
    5.资源释放
"""

import sys
import rclpy
from rclpy.node import Node
from rclpy.logging import get_logger
from base_interfaces_demo.srv import AddInts


class AddIntsClient(Node):
    def __init__(self):
        super().__init__("add_ints_client_node_py")

        # 3-1.创建客户端
        self.client = self.create_client(AddInts, "add_ints")
        self.get_logger().info("客户端已创建")

        self.future = None

        # 3-2.连接服务器
        self.connect_server()

    def connect_server(self):
        while not self.client.wait_for_service(2.0):
            self.get_logger().info("服务连接中...")

    def send_request(self):
        request = AddInts.Request()
        request.num1 = int(sys.argv[1])
        request.num2 = int(sys.argv[2])
        # 3-3.发送请求
        self.future = self.client.call_async(request)


def main():

    # 校验操作
    if len(sys.argv) != 3:
        get_logger("rclpy").error("请提交两个整型数据!")
        return 1

    rclpy.init()

    client = AddIntsClient()  # 4-1.创建客户端对象与连接服务
    client.send_request()  # 4-2.发送请求

    # 4-3.处理响应
    rclpy.spin_until_future_complete(client, client.future)
    try:
        response = client.future.result()
        client.get_logger().info(f"响应结果:{response.sum}")
    except Exception as e:
        client.get_logger().error(f"服务响应失败,异常信息为:{e}")

    rclpy.shutdown()


if __name__ == '__main__':
    main()
  • 修改　setup.py文件
### .....（原来）
    entry_points={
        'console_scripts': [
            'demo01_server = py02_service.demo01_server:main'
        ],
    },
  .....
### 修改之后
    entry_points={
        'console_scripts': [
            'demo01_server = py02_service.demo01_server_py:main',
            'demo02_client = py02_service.demo02_client_py:main'
        ],
    },
  • 编译

然后在进入ws01_plumbing工作空间，重新编译下 py02_service

colcon build --packages-select py02_service

-　运行测试

### 先启动服务端，等输出内容
source ./install/setup.bash
ros2 run py02_service demo01_server


### 启动client端，并且传入２个计算参数，等待结果
source ./install/setup.bash
ros2 run py02_service demo02_client 20 30

相关实现效果如下所示：

在这里插入图片描述

2.３ 动作通讯

2.3.1 场景

关于action通信，我们先从之前导航中的应用场景开始介绍，描述如下：

机器人导航到某个目标点,此过程需要一个节点A发布目标信息，然后一个节点B接收到请求并控制移动，最终响应目标达成状态信息。

乍一看，这好像是服务通信实现，因为需求中要A发送目标，B执行并返回结果，这是一个典型的基于请求响应的应答模式，不过，如果只是使用基本的服务通信实现，存在一个问题：导航是一个过程，是耗时操作，如果使用服务通信，那么只有在导航结束时，才会产生响应结果，而在导航过程中，节点A是不会获取到任何反馈的，从而可能出现程序"假死"的现象，过程的不可控意味着不良的用户体验，以及逻辑处理的缺陷(比如:导航中止的需求无法实现)。更合理的方案应该是:导航过程中，可以连续反馈当前机器人状态信息，当导航终止时，再返回最终的执行结果。在ROS中，该实现策略称之为：action 通信。

2.3.2 概念

动作通信适用于长时间运行的任务。就结构而言动作通信由目标、反馈和结果三部分组成；就功能而言动作通信类似于服务通信，动作客户端可以发送请求到动作服务端，并接收动作服务端响应的最终结果，不过动作通信可以在请求响应过程中获取连续反馈，并且也可以向动作服务端发送任务取消请求；就底层实现而言动作通信是建立在话题通信和服务通信之上的，目标发送实现是对服务通信的封装，结果的获取也是对服务通信的封装，而连续反馈则是对话题通信的封装。

2.3.3 作用

一般适用于耗时的请求响应场景，用以获取连续的状态反馈。

2.3.4 案例以及案例分析

准备

1.案例需求
需求：编写动作通信，动作客户端提交一个整型数据N，动作服务端接收请求数据并累加1-N之间的所有整数，将最终结果返回给动作客户端，且每累加一次都需要计算当前运算进度并反馈给动作客户端。

x

2.案例分析
在上述案例中，需要关注的要素有三个：

动作客户端；
动作服务端；
消息载体。

3.流程简介
案例实现前需要先自定义动作接口，接口准备完毕后，动作通信实现主要步骤如下：

编写动作服务端实现；
编写动作客户端实现；
编辑配置文件；
编译；
执行。

接下来，我们可以参考案例编写一个action文件，该文件中包含请求数据(一个整型字段)、响应数据(一个整型字段)和连续反馈数据(一个浮点型字段)。

步骤１：创建功能包

我们直接在上面ws01_plumbing/src文件夹下创建新的服务通信内容:

# 注意：相关依赖有rclpy & base_interfaces_demo
ros2 pkg create py03_action --build-type ament_python --dependencies rclpy base_interfaces_demo --node-name demo01_action_server
步骤２创建服务通信接口消息＆编辑相关文件
  • 1.创建编辑action文件　＆　修改配置文件
    在这里插入图片描述

2.编辑配置文件
package.xml
如果单独构建action功能包，需要在package.xml中需要添加一些依赖包.
当前使用的是 base_interfaces_demo 功能包，已经为 msg 、srv 文件添加过了一些依赖，所以 package.xml 中添加如下内容即可：

在这里插入图片描述
在这里插入图片描述

＃完整的所有的package.xml文件（msg, srv, action文件）如下：

<?xml version="1.0"?>
<?xml-model href="http://download.ros.org/schema/package_format3.xsd" schematypens="http://www.w3.org/2001/XMLSchema"?>
<package format="3">
  <name>base_interfaces_demo</name>
  <version>0.1.0</version>
  <description>存储各种通信接口定义文件的辅助功能包</description>
  <maintainer email="muzi2001@foxmail.com">muzing</maintainer>
  <license>TODO: License declaration</license>

  <buildtool_depend>ament_cmake</buildtool_depend>
  <buildtool_depend>rosidl_default_generators</buildtool_depend>

  <depend>action_msgs</depend>

  <!-- 编译依赖 -->
  <build_depend>rosidl_default_generators</build_depend>

  <test_depend>ament_lint_auto</test_depend>
  <test_depend>ament_lint_common</test_depend>

  <!-- 执行依赖 -->
  <exec_depend>rosidl_default_runtime</exec_depend>

  <!-- 声明当前包所属的功能包组 -->
  <member_of_group>rosidl_interface_packages</member_of_group>

  <export>
    <build_type>ament_cmake</build_type>
  </export>
</package>

2.CMakeLists.txt
如果是新建的功能包，与之前定义msg、srv文件同理，为了将文件转换成对应的C++和Python代码，还需要在CMakeLists.txt 中添加如下配置：

find_package(rosidl_default_generators REQUIRED)
# 为接口文件生成源码
rosidl_generate_interfaces(${PROJECT_NAME}
  "msg/Student.msg"
  "srv/AddInts.srv"
  "action/Progress.action"
)

不过，我们当前使用的base_interfaces_demo包，那么只需要修改rosidl_generate_interfaces函数即可，修改后的内容如下：
在这里插入图片描述

3.编译
终端中进入当前工作空间（ws01_plumbing），编译功能包：

colcon build --packages-select base_interfaces_demo

4.测试
编译完成之后，在工作空间下的 install 目录下将生成文件对应的C++和Python文件，我们也可以在终端下进入工作空间，通过如下命令查看文件定义以及编译是否正常：

正常情况下，终端将会输出与文件一致的内容。

ros2 interface show base_interfaces_demo/action/Progress 

int64 num
---
int64 sum
---
float32 progress
步骤３action通讯实现(python)
  • 修改server, client相关代码
# demo01_action_server.py内容
"""
需求:编写动作服务端,需要解析客户端提交的数字,遍历该数字并累加求和,将最终结果响应回
    客户端,且请求响应过程中需要生成连续反馈
流程:
    2.初始化ROS2客户端
    3.自定义节点类
        3-1.创建动作服务端对象
        3-2.处理提交的目标值(回调函数) --- 已有默认实现
        3-3.处理取消请求(回调函数) --- 已有默认实现
        3-4.生成连续反馈与最终响应(回调函数)
    4.调用spin函数
    5.资源释放
"""

import time

import rclpy
from rclpy.action import ActionServer
from rclpy.action.server import CancelResponse, GoalResponse, ServerGoalHandle
from rclpy.node import Node

from base_interfaces_demo.action import Progress


class ProgressActionServer(Node):
    def __init__(self):
        super().__init__("progress_action_sever_node_py")

        self.get_logger().info("动作通信服务端已创建")

        # 3-1.创建动作服务端对象
        self.server = ActionServer(
            self,
            Progress,
            "get_sum",
            self.execute_callback,
            goal_callback=self.goal_callback,
            cancel_callback=self.cancel_callback,
        )

    # 3-2.处理提交的目标值
    def goal_callback(self, goal_request):
        if goal_request.num > 1:
            return GoalResponse.ACCEPT
        else:
            self.get_logger().info("收到非法请求,已拒绝")
            return GoalResponse.REJECT

    # 3-3.处理取消请求(回调函数)
    def cancel_callback(self, cancel_request):
        self.get_logger().info("用户中止！")
        return CancelResponse.ACCEPT

    def execute_callback(self, goal_handle: ServerGoalHandle):
        # 3-4-1.生成连续反馈
        num = goal_handle.request.num
        sum = 0
        for i in range(1, num + 1):
            sum += i
            feedback = Progress.Feedback()
            feedback.progress = i / num
            goal_handle.publish_feedback(feedback)
            self.get_logger().info(f"连续反馈中,进度{feedback.progress*100:.1f}%")

            # 处理中止
            if goal_handle.is_cancel_requested:
                goal_handle.canceled()
                return Progress.Result()

            time.sleep(0.5)

        # 3-4-2.响应最终结果
        goal_handle.succeed()
        result = Progress.Result()
        result.sum = sum
        self.get_logger().info(f"最终计算结果为：{sum}")

        return result


def main(args=None):
    rclpy.init(args=args)
    rclpy.spin(ProgressActionServer())
    rclpy.shutdown()


if __name__ == '__main__':
    main()
  • 修改setup.py文件
from setuptools import find_packages, setup

package_name = 'py03_action'

setup(
    name=package_name,
    version='0.0.0',
    packages=find_packages(exclude=['test']),
    data_files=[
        ('share/ament_index/resource_index/packages',
            ['resource/' + package_name]),
        ('share/' + package_name, ['package.xml']),
    ],
    install_requires=['setuptools'],
    zip_safe=True,
    maintainer='fusionai',
    maintainer_email='ztfmars@163.com',
    description='TODO: Package description',
    license='TODO: License declaration',
    tests_require=['pytest'],
    entry_points={
        'console_scripts': [
            'demo01_action_server = py03_action.demo01_action_server:main',
            'demo02_action_client = py03_action.demo02_action_client:main',
        ],
    },
)
  • 编译

然后在进入ws01_plumbing工作空间，重新编译下 py0３_action

colcon build --packages-select py03_action

-　运行测试

### 先启动服务端，等输出内容
source ./install/setup.bash
ros2 run py03_action demo01_action_server


### 启动client端，并且传入２个计算参数，等待结果
source ./install/setup.bash
ros2 run py03_action demo02_action_client 10

相关实现效果如下所示：

在这里插入图片描述

2.5 参数服务

2.5.1 场景

在机器人系统中不同的功能模块可能会使用到一些相同的数据，比如：

导航实现时，会进行路径规划，路径规划主要包含， 全局路径规划和本地路径规划，所谓全局路径规划就是设计一个从出发点到目标点的大致路径，
而本地路径规划，则是根据车辆当前路况生成实时的行进路径。两种路径规划实现，
都会使用到车辆的尺寸数据——长度、宽度、高度等。那么这些通用数据在程序中应该如何存储、调用呢？

上述场景中，就可以使用参数服务实现，在一个节点下保存车辆尺寸数据，其他节点可以访问该节点并操作这些数据。

2.5.2 概念

参数服务是以共享的方式实现不同节点之间数据交互的一种通信模式。保存参数的节点称之为参数服务端，调用参数的节点称之为参数客户端。参数客户端与参数服务端的交互是基于请求响应的，且参数通信的实现本质上对服务通信的进一步封装。

2.5.3 作用

参数服务保存的数据类似于编程中“全局变量”的概念，可以在不同的节点之间共享数据。

2.5.4 案例以及案例分析

1.案例需求
需求：在参数服务端设置一些参数，参数客户端访问服务端并操作这些参数。

xx1

2.案例分析
在上述案例中，需要关注的要素有三个：

参数客户端；
参数服务端；
参数。

3.流程简介
案例实现前需要先了解ROS2中参数的相关API，无论是客户端还是服务端都会使用到参数，而参数服务案例实现主要步骤如下：

编写参数服务端实现；
编写参数客户端实现；
编辑配置文件；
编译；
执行。

说明
  • 参数数据类型

在ROS2中，参数由键、值和描述符三部分组成，其中键是字符串类型，值可以是bool、int64、float64、string、byte[]、bool[]、int64[]、float64[]、string[]中的任一类型，描述符默认情况下为空，但是可以设置参数描述、参数数据类型、取值范围或其他约束等信息。

为了方便操作，参数被封装为了相关类，其中C++客户端对应的类是，Python客户端对应的类是。借助于相关API，我们可以实现参数对象创建以及参数属性解析等操作。以下代码提供了参数相关API基本使用的示例。

1.2

1.3

2.5.5 整体步骤

步骤１： 创建python 参数包功能包
cd ws01_plumbing/src
ros2 pkg create py04_param --build-type ament_python --dependencies rclpy --node-name demo00_param
步骤２：客户端／服务端相关代码
## demo00_param.py
"""
1.创建参数对象
2.解析参数
"""

import rclpy
from rclpy.node import Node


class MyParam(Node):
    def __init__(self):
        super().__init__("mu_param_node")
        self.get_logger().info("参数API使用")

        # 创建参数对象
        p1 = rclpy.Parameter("car_name", value="Tiger")
        p2 = rclpy.Parameter("width", value=1.5)
        p3 = rclpy.Parameter("wheels", value=2)

        # 解析参数
        self.get_logger().info(f"car_name = {p1.value}")
        self.get_logger().info(f"width = {p2.value}")
        self.get_logger().info(f"wheels = {p3.value}")

        self.get_logger().info(f"key = {p1.name}")


def main():
    rclpy.init()
    MyParam()
    rclpy.shutdown()


if __name__ == '__main__':
    main()

### demo01_param_server.py
"""
需求：创建参数服务端，并操作参数（增删改查）
流程：
    自定义节点类
    1.增
    2.查
    3.改
    4.删
"""

import rclpy
from rclpy.node import Node


class ParamServer(Node):
    def __init__(self):
        # 如果允许删除参数，那么需要提前声明
        super().__init__("param_server_node_py", allow_undeclared_parameters=True)
        self.get_logger().info("参数服务端已创建")

    # 1.增
    def declare_param(self):
        self.get_logger().info("----------新增参数----------")
        self.declare_parameter("car_name", "tiger")
        self.declare_parameter("width", 1.55)
        self.declare_parameter("wheels", 5)

        # 设置未声明的参数，前提：allow_undeclared_parameters=True
        self.set_parameters([rclpy.Parameter("height", value=1.6)])

    # 2.查
    def get_param(self):
        self.get_logger().info("----------查询参数----------")

        # 获取指定参数
        car_name = self.get_parameter("car_name")
        self.get_logger().info(f"{car_name.name} = {car_name.value}")

        # 获取多个参数
        params = self.get_parameters(["car_name", "wheels", "width", "height"])
        for par in params:
            self.get_logger().info(f"{par.name} == {par.value}")

        # 判断是否包含某个参数
        self.get_logger().info(f"是否包含car_name: {self.has_parameter('car_name')}")
        self.get_logger().info(f"是否包含height: {self.has_parameter('height')}")

    # 3.改
    def update_param(self):
        self.get_logger().info("----------修改参数----------")
        self.set_parameters([rclpy.Parameter("car_name", value="dragon")])
        car_name = self.get_parameter("car_name")
        self.get_logger().info(f"修改后{car_name.name} = {car_name.value}")

    # 4.删
    def del_param(self):
        self.get_logger().info("----------删除参数----------")
        self.undeclare_parameter("car_name")  # 可以删除通过declare_param设置的参数值
        self.get_logger().info(f"是否包含car_name: {self.has_parameter('car_name')}")


def main():
    rclpy.init()

    test_node = ParamServer()
    test_node.declare_param()
    test_node.get_param()
    test_node.update_param()
    test_node.del_param()
    rclpy.spin(test_node)

    rclpy.shutdown()


if __name__ == '__main__':
    main()

修改setup.py文件

    entry_points={
        'console_scripts': [
            'demo00_param = py04_param.demo00_param:main',
            'demo01_param_server = py04_param.demo01_param_server:main',
        ],
    },
步骤３：编译和执行

然后在进入ws01_plumbing工作空间，重新编译下 py04_param

colcon build --packages-select py04_param

运行测试

### 先启动服务端，等输出内容
source ./install/setup.bash
ros2 run py04_param demo00_param


### 启动client端，并且传入２个计算参数，等待结果
source ./install/setup.bash
ros2 run py04_param demo01_param_server

注意：如何单独显示或者获取相关内容，可以使用list 或者get指令获取

在这里插入图片描述

相关实现效果如下所示：

在这里插入图片描述

3.通讯内容总结

在这里插入图片描述

ref

ROS2快速使用博客
ROS2理论与实践 1-3章
ROS2理论与实践第4-6章
ROS2理论与实践第8-9章
赵虚左Ros2-核心篇讲义学习-第二三章 ROS2通信机制核心
赵虚左Ros2-服务通信

!!!相关所有源码内容，重点参见如下链接!!!
ROS2_Learning Public相关源码内容

# 机器人
Logo
全球具身智能开发者社区

立足具身智能前沿赛道，致力于搭建全球化、开源化、全栈式技术交流与实践共创平台。

更多推荐

cover

Mac 用户如何玩转 LeRobot？M3 芯片部署6 轴机械臂SoArm101经验总结 第3篇

avatar 全球具身智能开发者社区
cover

2026世界人工智能大会亮点多：AI产业走向现实交付，多款创新产品亮相！

avatar 全球具身智能开发者社区
cover

2026具身智能迈向场景落地期，库萨科技全栈技术能否打通城市服务机器人规模化之路？

avatar 全球具身智能开发者社区