话题：节点间传递数据的桥梁

通信模式

以两个机器人节点为例。A节点的功能是驱动相机这个硬件设备，获取得到相机拍摄的图像信息，B节点的功能是视频监控，将相机拍摄到的图像实时显示给用户查看。

让我们思考一下，这两个节点之间是否存在必然联系？显然，节点A需要向节点B传输图像数据，因为只有获取到数据后，节点B才能进行可视化渲染。

在ROS框架中，这种节点间的数据传输机制被称为"话题"。话题就像一座桥梁，专门负责在特定方向上实现节点之间的数据传递。

发布/订阅模型

从话题本身的实现角度来看，使用了基于DDS的发布/订阅模型，什么叫发布和订阅呢？

话题数据传输的特性是从一个节点到另外一个节点，发送数据的对象称之为发布者，接收数据的对象称之为订阅者，每一个话题都需要有一个名字，传输的数据也需要有固定的数据类型。

举个例子，就像我们平时关注微信公众号一样。以"AAA"这个公众号为例："AAA"就是话题名称，运营者作为发布者将整理好的机器人知识编辑成标准格式的文章发布。这种文章格式就是话题的数据类型。感兴趣的用户可以订阅该公众号，订阅后就能接收到推送内容，否则就无法获取信息。

这种发布/订阅模式在生活中很常见，比如订阅报纸、杂志等都是类似的场景。

多对多通信

让我们仔细思考这些可订阅的内容——它们并不是唯一的。每个人可以同时订阅多个公众号、报纸和杂志，而这些媒体也可以被众多用户订阅。同理，在ROS中，话题的发布者和订阅者数量同样不受限制，这种通信模式被称为多对多模型。

由于话题采用多对多机制，控制指令的发布源可以是单个摇杆，也可以是多个；接收指令的机器人数量同样可以灵活配置。想象一下这个场景：多个控制节点同时向一群机器人发送指令——虽然画面颇具戏剧性，但需要注意为不同指令源设置优先级，否则机器人可能会陷入"不知道该听谁指挥"的混乱状态。

异步通讯

通信还有一个重要特性：异步。可能有些同学是第一次接触这个概念。所谓异步，指的是发布者发送数据后，并不清楚订阅者何时会收到。就像AAA公众号发布文章时，无法知道读者具体何时阅读；或者报社发行报纸时，也无法确定读者何时收到。这就是异步的本质。

异步特性使得话题通信特别适合处理周期性发布的数据，例如传感器数据和运动控制指令。但对于一些需要即时响应的逻辑性指令（比如参数修改），话题通信就不太适用了。

消息接口

最后，为了实现高效的数据传输，发布者和订阅者必须使用统一的数据描述格式——就好比双方需要使用相同的语言才能准确沟通。在ROS中，这种标准化格式被称为"消息"(message)，它相当于编程语言中的数据结构。以图像数据为例，消息会明确定义图像的宽度、高度等像素信息，以及每个像素的RGB值等属性，这些在ROS中都有统一规范。

消息作为ROS的核心接口机制，采用与编程语言无关的.msg文件格式进行定义。开发者可以根据需求自定义消息类型。这种灵活的接口设计让各个功能节点能够像积木一样，通过标准化的接口自由组合，从而构建出复杂的机器人系统。

案例一：Hello World话题通信

还是从Hello World例程开始，我们来创建一个发布者，发布话题“chatter”，周期发送“Hello World”这个字符串，消息类型是ROS中标准定义的String，再创建一个订阅者，订阅“chatter”这个话题，从而接收到“Hello World”这个字符串。

运行效果

启动第一个终端，运行话题的发布者节点：

ros2 run learning_topic topic_helloworld_pub

启动第二个终端，运行话题的订阅者节点：

ros2 run learning_topic topic_helloworld_sub

可以看到发布者循环发布“Hello World”字符串消息，订阅者也以几乎同样的频率收到该话题的消息数据。

发布者代码解析

import rclpy                                     # ROS2 Python接口库
from rclpy.node import Node                      # ROS2 节点类
from std_msgs.msg import String                  # 字符串消息类型

"""
创建一个发布者节点
"""
class PublisherNode(Node):

    def __init__(self, name):
        super().__init__(name)                                    # ROS2节点父类初始化
        self.pub = self.create_publisher(String, "chatter", 10)   # 创建发布者对象（消息类型、话题名、队列长度）
        self.timer = self.create_timer(0.5, self.timer_callback)  # 创建一个定时器（单位为秒的周期，定时执行的回调函数）

    def timer_callback(self):                                     # 创建定时器周期执行的回调函数
        msg = String()                                            # 创建一个String类型的消息对象
        msg.data = 'Hello World'                                  # 填充消息对象中的消息数据
        self.pub.publish(msg)                                     # 发布话题消息
        self.get_logger().info('Publishing: "%s"' % msg.data)     # 输出日志信息，提示已经完成话题发布

def main(args=None):                                 # ROS2节点主入口main函数
    rclpy.init(args=args)                            # ROS2 Python接口初始化
    node = PublisherNode("topic_helloworld_pub")     # 创建ROS2节点对象并进行初始化
    rclpy.spin(node)                                 # 循环等待ROS2退出
    node.destroy_node()                              # 销毁节点对象
    rclpy.shutdown()                                 # 关闭ROS2 Python接口

完成代码的编写后需要设置功能包的编译选项，让系统知道Python程序的入口，打开功能包的setup.py文件，加入如下入口点的配置：

 entry_points={
        'console_scripts': [
         'topic_helloworld_pub  = learning_topic.topic_helloworld_pub:main',
        ],
    },

代码链路

功能

创建 ROS2 节点，每 0.5 秒向 /chatter 话题发布 Hello World 字符串。

核心四步（ROS2 固定模板）
  1. 初始化环境：rclpy.init()
  2. 创建节点：继承 Node，注册发布者、定时器
  3. 自旋运行：rclpy.spin() 阻塞监听，自动执行定时器回调、发布消息
  4. 释放资源：销毁节点 + rclpy.shutdown()
关键代码核心作用
  • create_publisher(String, "chatter", 10)：创建话题发布器，指定消息类型、话题名、消息队列
  • create_timer(0.5, 回调)：定时 0.5s 触发，自动执行发布逻辑
  • msg.data：ROS 消息固定数据字段，赋值后即可发布
  • get_logger().info()：ROS 规范日志输出
最简执行链路

初始化 ROS 环境 → 新建节点 (绑定发布器 + 定时器) → 持续自旋发布消息 → 退出后清理资源

订阅者代码解析

import rclpy                      # ROS2 Python接口库
from rclpy.node   import Node     # ROS2 节点类
from std_msgs.msg import String   # ROS2标准定义的String消息

"""
创建一个订阅者节点
"""
class SubscriberNode(Node):

    def __init__(self, name):
        super().__init__(name)                             # ROS2节点父类初始化
        self.sub = self.create_subscription(\
            String, "chatter", self.listener_callback, 10) # 创建订阅者对象（消息类型、话题名、订阅者回调函数、队列长度）

    def listener_callback(self, msg):                      # 创建回调函数，执行收到话题消息后对数据的处理
        self.get_logger().info('I heard: "%s"' % msg.data) # 输出日志信息，提示订阅收到的话题消息

def main(args=None):                               # ROS2节点主入口main函数
    rclpy.init(args=args)                          # ROS2 Python接口初始化
    node = SubscriberNode("topic_helloworld_sub")  # 创建ROS2节点对象并进行初始化
    rclpy.spin(node)                               # 循环等待ROS2退出
    node.destroy_node()                            # 销毁节点对象
    rclpy.shutdown()                               # 关闭ROS2 Python接口

完成代码的编写后需要设置功能包的编译选项，让系统知道Python程序的入口，打开功能包的setup.py文件，加入如下入口点的配置：

 entry_points={
        'console_scripts': [
         'topic_helloworld_pub  = learning_topic.topic_helloworld_pub:main',
         'topic_helloworld_sub  = learning_topic.topic_helloworld_sub:main',
        ],
    },

代码链路

功能

订阅话题 chatter，收到消息自动执行回调打印内容。

ROS2 固定生命周期模板（和发布者一致）

  1. rclpy.init()  初始化 ROS2 上下文
  2. 实例化节点  内部创建订阅器
  3. rclpy.spin(node) 自旋阻塞，驱动回调执行（不 spin 收不到消息）
  4. destroy_node() 销毁节点
  5. rclpy.shutdown() 关闭 ROS2

核心 API

# 创建订阅
self.create_subscription(String, "chatter", self.listener_callback, 10)
  • 消息类型：String
  • 话题名：chatter（必须和发布者完全一致）
  • 回调：收到消息自动执行，参数msg为接收的消息对象
  • 队列：10
# 消息回调
def listener_callback(self, msg):
    self.get_logger().info(f'I heard: "{msg.data}"')

msg.data 获取字符串数据。

setup.py 入口

entry_points={
    'console_scripts': [
        'topic_helloworld_sub = learning_topic.topic_helloworld_sub:main',
    ]
}

运行：

ros2 run learning_topic topic_helloworld_sub

数据流

发布者 publish → DDS → 订阅者触发 listener_callback → 业务处理

案例二：机器视觉识别

在节点概念的讲解过程中，我们通过一个节点驱动了相机，并且实现了对红色物体的识别。功能虽然没问题，但是对于机器人开发来讲，并没有做到程序的模块化，更好的方式是将相机驱动和视觉识别做成两个节点，节点间的联系就是这个图像数据，通过话题周期传输即可。

运行效果

这个图像消息在ROS中是标准定义好的，如果未来要更换另一个相机，只需要修改驱动节点，视觉识别节点完全是软件功能，就可以保持不变了，这种模块化的设计思想，可以更好的保证软件的可移植性。

好啦，说干就干，我们先来看下效果如何？

启动两个终端，分别运行以下两个节点，第一个节点驱动相机并发布图像话题，第二个节点订阅图像话题并实现视觉识别。

$ ros2 run learning_topic topic_webcam_pub
$ ros2 run learning_topic topic_webcam_sub

将红色物体放入相机范围内，即可看到识别效果。

发布者代码解析

import rclpy                        # ROS2 Python接口库
from rclpy.node import Node         # ROS2 节点类
from sensor_msgs.msg import Image   # 图像消息类型
from cv_bridge import CvBridge      # ROS与OpenCV图像转换类
import cv2                          # Opencv图像处理库

"""
创建一个发布者节点
"""
class ImagePublisher(Node):

    def __init__(self, name):
        super().__init__(name)                                           # ROS2节点父类初始化
        self.publisher_ = self.create_publisher(Image, 'image_raw', 10)  # 创建发布者对象（消息类型、话题名、队列长度）
        self.timer = self.create_timer(0.1, self.timer_callback)         # 创建一个定时器（单位为秒的周期，定时执行的回调函数）
        self.cap = cv2.VideoCapture(0)                                   # 创建一个视频采集对象，驱动相机采集图像（相机设备号）
        self.cv_bridge = CvBridge()                                      # 创建一个图像转换对象，用于稍后将OpenCV的图像转换成ROS的图像消息

    def timer_callback(self):
        ret, frame = self.cap.read()                         # 一帧一帧读取图像

        if ret == True:                                      # 如果图像读取成功
            self.publisher_.publish(
                self.cv_bridge.cv2_to_imgmsg(frame, 'bgr8')) # 发布图像消息

        self.get_logger().info('Publishing video frame')     # 输出日志信息，提示已经完成图像话题发布

def main(args=None):                                 # ROS2节点主入口main函数
    rclpy.init(args=args)                            # ROS2 Python接口初始化
    node = ImagePublisher("topic_webcam_pub")        # 创建ROS2节点对象并进行初始化
    rclpy.spin(node)                                 # 循环等待ROS2退出
    node.destroy_node()                              # 销毁节点对象
    rclpy.shutdown()                                 # 关闭ROS2 Python接口

完成代码的编写后需要设置功能包的编译选项，让系统知道Python程序的入口，打开功能包的setup.py文件，加入如下入口点的配置：

   entry_points={
        'console_scripts': [
         'topic_helloworld_pub  = learning_topic.topic_helloworld_pub:main',
         'topic_helloworld_sub  = learning_topic.topic_helloworld_sub:main',
         'topic_webcam_pub      = learning_topic.topic_webcam_pub:main',
        ],
    },

订阅者代码解析

learning_topic/topic_webcam_sub.py
 

import rclpy                            # ROS2 Python接口库
from rclpy.node import Node             # ROS2 节点类
from sensor_msgs.msg import Image       # 图像消息类型
from cv_bridge import CvBridge          # ROS与OpenCV图像转换类
import cv2                              # Opencv图像处理库
import numpy as np                      # Python数值计算库

lower_red = np.array([0, 90, 128])      # 红色的HSV阈值下限
upper_red = np.array([180, 255, 255])   # 红色的HSV阈值上限

"""
创建一个订阅者节点
"""
class ImageSubscriber(Node):
    def __init__(self, name):
        super().__init__(name)                                # ROS2节点父类初始化
        self.sub = self.create_subscription(
            Image, 'image_raw', self.listener_callback, 10)   # 创建订阅者对象（消息类型、话题名、订阅者回调函数、队列长度）
        self.cv_bridge = CvBridge()                           # 创建一个图像转换对象，用于OpenCV图像与ROS的图像消息的互相转换

    def object_detect(self, image):
        hsv_img = cv2.cvtColor(image, cv2.COLOR_BGR2HSV)      # 图像从BGR颜色模型转换为HSV模型
        mask_red = cv2.inRange(hsv_img, lower_red, upper_red) # 图像二值化
        contours, hierarchy = cv2.findContours(
            mask_red, cv2.RETR_LIST, cv2.CHAIN_APPROX_NONE)   # 图像中轮廓检测

        for cnt in contours:                                  # 去除一些轮廓面积太小的噪声
            if cnt.shape[0] < 150:
                continue

            (x, y, w, h) = cv2.boundingRect(cnt)              # 得到苹果所在轮廓的左上角xy像素坐标及轮廓范围的宽和高
            cv2.drawContours(image, [cnt], -1, (0, 255, 0), 2)# 将苹果的轮廓勾勒出来
            cv2.circle(image, (int(x+w/2), int(y+h/2)), 5,
                       (0, 255, 0), -1)                       # 将苹果的图像中心点画出来

        cv2.imshow("object", image)                           # 使用OpenCV显示处理后的图像效果
        cv2.waitKey(10)

    def listener_callback(self, data):
        self.get_logger().info('Receiving video frame')     # 输出日志信息，提示已进入回调函数
        image = self.cv_bridge.imgmsg_to_cv2(data, 'bgr8')  # 将ROS的图像消息转化成OpenCV图像
        self.object_detect(image)                           # 苹果检测


def main(args=None):                            # ROS2节点主入口main函数
    rclpy.init(args=args)                       # ROS2 Python接口初始化
    node = ImageSubscriber("topic_webcam_sub")  # 创建ROS2节点对象并进行初始化
    rclpy.spin(node)                            # 循环等待ROS2退出
    node.destroy_node()                         # 销毁节点对象
    rclpy.shutdown()                            # 关闭ROS2 Python接口

完成代码的编写后需要设置功能包的编译选项，让系统知道Python程序的入口，打开功能包的setup.py文件，加入如下入口点的配置：

entry_points={
        'console_scripts': [
         'topic_helloworld_pub  = learning_topic.topic_helloworld_pub:main',
         'topic_helloworld_sub  = learning_topic.topic_helloworld_sub:main',
         'topic_webcam_pub      = learning_topic.topic_webcam_pub:main',
         'topic_webcam_sub      = learning_topic.topic_webcam_sub:main',
        ],
    },

案例三：机器视觉识别优化

通过话题对原本节点功能的解耦，似乎让视觉识别的例程焕然一新了，不过似乎还有哪里不太对劲，大家有感觉到么？

ROS的目标不是提高软件复用率么，现在视觉识别的节点可以复用了，相机驱动节点好像不行呀，每换一个相机，是不是都得换一个驱动节点？这当然是不可能的！

常用的usb相机驱动一般都是通用的，ROS中也集成了usb相机的标准驱动，我们只需要通过这样一行指令，就可以安装好，无论你用什么样的相机，只要符合usb接口协议，就可以直接使用ROS中的相机驱动节点，发布标准的图像话题了。

$ sudo apt install ros-humble-usb-cam

这样，我们的代码又得到了进一步精简，刚才自己写的图像发布节点换成了这样一句指令，视觉识别节点不需要做任何变化。

$ ros2 run usb_cam usb_cam_node_exe
$ ros2 run learning_topic topic_webcam_sub

话题命令行操作

话题命令的常用操作如下：

$ ros2 topic list                # 查看话题列表
$ ros2 topic info <topic_name>   # 查看话题信息
$ ros2 topic hz <topic_name>     # 查看话题发布频率
$ ros2 topic bw <topic_name>     # 查看话题传输带宽
$ ros2 topic echo <topic_name>   # 查看话题数据
$ ros2 topic pub <topic_name> <msg_type> <msg_data>   # 发布话题消息

服务

话题通信可以实现多个ROS节点之间数据的单向传输，使用这种异步通信机制，发布者无法准确知道订阅者是否收到消息，本讲我们将一起学习ROS另外一种常用的通信方法——服务，可以实现类似你问我答的同步通信效果。

通信模型

在之前的课程中，我们通过一个节点驱动相机，发布图像话题，另外一个节点订阅图像话题，并实现对其中红色物体的识别，此时我们可以按照图像识别的频率，周期得到物体的位置。

该位置信息可进一步传输至机器人的上层应用，用于实现目标跟随或移动到指定位置附近的功能。此时，我们无需持续高频订阅物体位置数据，而是在需要时发送查询请求，即可快速获取目标的最新位置信息。

这种通信模式与单向话题传输不同，采用了"请求-应答"的交互方式。在ROS中，这种通信机制被称为服务（Service）。

客户端/服务器模型

从服务的实现机制上来看，这种你问我答的形式叫做客户端/服务器模型，简称为CS模型，客户端在需要某些数据的时候，针对某个具体的服务，发送请求信息，服务器端收到请求之后，就会进行处理并反馈应答信息。

这种通信机制在日常生活中随处可见。比如我们浏览网页时，电脑浏览器作为客户端，通过输入网址或点击链接向网站服务器发出请求，服务器接收后会返回相应的页面数据。

同步通信

这一过程通常要求尽可能快速响应。如果服务器长时间无反应，导致浏览器持续加载（转圈圈），则可能是服务器宕机或网络故障。与话题通信不同，在服务通信中，客户端能够通过接收到的应答信息来判断服务器状态，这种方式也被称为同步通信。

一对多通信

服务接口

在服务通信中，数据传递依然是核心功能。与话题通信不同的是，服务通信的数据被划分为两部分：请求数据（如获取苹果位置的指令）和响应数据（如返回的苹果坐标信息）。这些数据格式与话题消息类似，都需要在ROS中进行标准化定义：话题使用.msg文件定义，而服务则采用.srv文件定义。我们将在后续内容中详细介绍具体的定义方法。

案例一：加法求解器

大家现在对ROS服务通信应该有了基本了解，接下来我们就要开始编写代码啦。还是从一个相对简单的例程开始，也是ROS官方的一个例程，通过服务实现一个加法求解器的功能。

要计算两个数的和，客户端节点将这两个加数封装成请求数据，发送至"add_two_ints"服务。服务端节点收到请求后执行加法运算，并将结果封装成应答数据返回给客户端。这样，客户端就能获取所需的计算结果。

运行效果

我们一起操作下这个例程，并且看下代码的实现原理。

启动两个终端，并运行如下节点，第一个节点是服务端，等待请求数据并提供求和功能，第二个节点是客户端，发送传入的两个加数并等待求和结果。

$ ros2 run learning_service service_adder_server
$ ros2 run learning_service service_adder_client 2 3

服务端

客户端

客户端代码解析

import sys

import rclpy                                  # ROS2 Python接口库
from rclpy.node   import Node                 # ROS2 节点类
from learning_interface.srv import AddTwoInts # 自定义的服务接口

class adderClient(Node):
    def __init__(self, name):
        super().__init__(name)                                       # ROS2节点父类初始化
        self.client = self.create_client(AddTwoInts, 'add_two_ints') # 创建服务客户端对象（服务接口类型，服务名）
        while not self.client.wait_for_service(timeout_sec=1.0):     # 循环等待服务器端成功启动
            self.get_logger().info('service not available, waiting again...') 
        self.request = AddTwoInts.Request()                          # 创建服务请求的数据对象

    def send_request(self):                                          # 创建一个发送服务请求的函数
        self.request.a = int(sys.argv[1])
        self.request.b = int(sys.argv[2])
        self.future = self.client.call_async(self.request)           # 异步方式发送服务请求

def main(args=None):
    rclpy.init(args=args)                        # ROS2 Python接口初始化
    node = adderClient("service_adder_client")   # 创建ROS2节点对象并进行初始化
    node.send_request()                          # 发送服务请求

    while rclpy.ok():                            # ROS2系统正常运行
        rclpy.spin_once(node)                    # 循环执行一次节点

        if node.future.done():                   # 数据是否处理完成
            try:
                response = node.future.result()  # 接收服务器端的反馈数据
            except Exception as e:
                node.get_logger().info(
                    'Service call failed %r' % (e,))
            else:
                node.get_logger().info(          # 将收到的反馈信息打印输出
                    'Result of add_two_ints: for %d + %d = %d' % 
                    (node.request.a, node.request.b, response.sum))
            break

    node.destroy_node()                          # 销毁节点对象
    rclpy.shutdown()                             # 关闭ROS2 Python接口

完成代码的编写后需要设置功能包的编译选项，让系统知道Python程序的入口，打开功能包的setup.py文件，加入如下入口点的配置：

  entry_points={
        'console_scripts': [
         'service_adder_client  = learning_service.service_adder_client:main',
        ],
    },
流程总结

对以上程序进行分析，如果我们想要实现一个客户端，流程如下：

  • 编程接口初始化
  • 创建节点并初始化
  • 创建客户端对象
  • 创建并发送请求数据
  • 等待服务器端应答数据
  • 销毁节点并关闭接口

服务端代码解析

至于服务器端的实现，有点类似话题通信中的订阅者，并不知道请求数据什么时间出现，也用到了回调函数机制。

程序实现
import rclpy                                     # ROS2 Python接口库
from rclpy.node   import Node                    # ROS2 节点类
from learning_interface.srv import AddTwoInts    # 自定义的服务接口

class adderServer(Node):
    def __init__(self, name):
        super().__init__(name)                                                           # ROS2节点父类初始化
        self.srv = self.create_service(AddTwoInts, 'add_two_ints', self.adder_callback)  # 创建服务器对象（接口类型、服务名、服务器回调函数）

    def adder_callback(self, request, response):   # 创建回调函数，执行收到请求后对数据的处理
        response.sum = request.a + request.b       # 完成加法求和计算，将结果放到反馈的数据中
        self.get_logger().info('Incoming request\na: %d b: %d' % (request.a, request.b))   # 输出日志信息，提示已经完成加法求和计算
        return response                          # 反馈应答信息

def main(args=None):                             # ROS2节点主入口main函数
    rclpy.init(args=args)                        # ROS2 Python接口初始化
    node = adderServer("service_adder_server")   # 创建ROS2节点对象并进行初始化
    rclpy.spin(node)                             # 循环等待ROS2退出
    node.destroy_node()                          # 销毁节点对象
    rclpy.shutdown()                             # 关闭ROS2 Python接口

完成代码的编写后需要设置功能包的编译选项，让系统知道Python程序的入口，打开功能包的setup.py文件，加入如下入口点的配置：

    entry_points={
        'console_scripts': [
         'service_adder_client  = learning_service.service_adder_client:main',
         'service_adder_server  = learning_service.service_adder_server:main',
        ],
    },
流程总结

对以上程序进行分析，如果我们想要实现一个服务端，流程如下：

  • 编程接口初始化
  • 创建节点并初始化
  • 创建服务器端对象
  • 通过回调函数处进行服务
  • 向客户端反馈应答结果
  • 销毁节点并关闭接口

案例二：机器视觉识别

三个节点出现

  1. 相机驱动节点，发布图像数据；
  2. 视觉识别节点，订阅图像数据，并且集成了一个服务器端对象，随时准备提供目标位置；
  3. 客户端节点，我们可以认为是一个机器人目标跟踪的节点，当需要根据目标运动时，就发送一次请求，然后拿到一个当前的目标位置。

启动三个终端：

ros2 run usb_cam usb_cam_node_exe
ros2 run learning_service service_object_server
ros2 run learning_service service_object_client

相机驱动节点:

视觉识别节点:
 




客户端代码解析


import rclpy                                            # ROS2 Python接口库
from rclpy.node   import Node                           # ROS2 节点类
from learning_interface.srv import GetObjectPosition    # 自定义的服务接口

class objectClient(Node):
    def __init__(self, name):
        super().__init__(name)                          # ROS2节点父类初始化
        self.client = self.create_client(GetObjectPosition, 'get_target_position')
        while not self.client.wait_for_service(timeout_sec=1.0):
            self.get_logger().info('service not available, waiting again...')
        self.request = GetObjectPosition.Request()

    def send_request(self):
        self.request.get = True
        self.future = self.client.call_async(self.request)

def main(args=None):
    rclpy.init(args=args)                             # ROS2 Python接口初始化
    node = objectClient("service_object_client")      # 创建ROS2节点对象并进行初始化
    node.send_request()

    while rclpy.ok():
        rclpy.spin_once(node)

        if node.future.done():
            try:
                response = node.future.result()
            except Exception as e:
                node.get_logger().info(
                    'Service call failed %r' % (e,))
            else:
                node.get_logger().info(
                    'Result of object position:\n x: %d y: %d' %
                    (response.x, response.y))
            break
    node.destroy_node()                              # 销毁节点对象
    rclpy.shutdown()                                 # 关闭ROS2 Python接口

完成代码的编写后需要设置功能包的编译选项，让系统知道Python程序的入口，打开功能包的setup.py文件，加入如下入口点的配置：

  entry_points={
        'console_scripts': [
         'service_adder_client  = learning_service.service_adder_client:main',
         'service_adder_server  = learning_service.service_adder_server:main',
         'service_object_client = learning_service.service_object_client:main',
        ],
    },

服务端代码解析

import rclpy                                           # ROS2 Python接口库
from rclpy.node import Node                            # ROS2 节点类
from sensor_msgs.msg import Image                      # 图像消息类型
import numpy as np                                     # Python数值计算库
from cv_bridge import CvBridge                         # ROS与OpenCV图像转换类
import cv2                                             # Opencv图像处理库
from learning_interface.srv import GetObjectPosition   # 自定义的服务接口

lower_red = np.array([0, 90, 128])     # 红色的HSV阈值下限
upper_red = np.array([180, 255, 255])  # 红色的HSV阈值上限

class ImageSubscriber(Node):
    def __init__(self, name):
        super().__init__(name)                              # ROS2节点父类初始化
        self.sub = self.create_subscription(
            Image, 'image_raw', self.listener_callback, 10) # 创建订阅者对象（消息类型、话题名、订阅者回调函数、队列长度）
        self.cv_bridge = CvBridge()                         # 创建一个图像转换对象，用于OpenCV图像与ROS的图像消息的互相转换

        self.srv = self.create_service(GetObjectPosition,   # 创建服务器对象（接口类型、服务名、服务器回调函数）
                                       'get_target_position',
                                       self.object_position_callback)    
        self.objectX = 0
        self.objectY = 0                              

    def object_detect(self, image):
        hsv_img = cv2.cvtColor(image, cv2.COLOR_BGR2HSV)      # 图像从BGR颜色模型转换为HSV模型
        mask_red = cv2.inRange(hsv_img, lower_red, upper_red) # 图像二值化
        contours, hierarchy = cv2.findContours(
            mask_red, cv2.RETR_LIST, cv2.CHAIN_APPROX_NONE)   # 图像中轮廓检测

        for cnt in contours:                                  # 去除一些轮廓面积太小的噪声
            if cnt.shape[0] < 150:
                continue

            (x, y, w, h) = cv2.boundingRect(cnt)              # 得到苹果所在轮廓的左上角xy像素坐标及轮廓范围的宽和高
            cv2.drawContours(image, [cnt], -1, (0, 255, 0), 2)# 将苹果的轮廓勾勒出来
            cv2.circle(image, (int(x+w/2), int(y+h/2)), 5,
                       (0, 255, 0), -1)                       # 将苹果的图像中心点画出来

            self.objectX = int(x+w/2)
            self.objectY = int(y+h/2)

        cv2.imshow("object", image)                            # 使用OpenCV显示处理后的图像效果
        cv2.waitKey(50)

    def listener_callback(self, data):
        self.get_logger().info('Receiving video frame')        # 输出日志信息，提示已进入回调函数
        image = self.cv_bridge.imgmsg_to_cv2(data, 'bgr8')     # 将ROS的图像消息转化成OpenCV图像
        self.object_detect(image)                              # 苹果检测

    def object_position_callback(self, request, response):     # 创建回调函数，执行收到请求后对数据的处理
        if request.get == True:
            response.x = self.objectX                          # 目标物体的XY坐标
            response.y = self.objectY
            self.get_logger().info('Object position\nx: %d y: %d' %
                                   (response.x, response.y))   # 输出日志信息，提示已经反馈
        else:
            response.x = 0
            response.y = 0
            self.get_logger().info('Invalid command')          # 输出日志信息，提示已经反馈
        return response


def main(args=None):                                 # ROS2节点主入口main函数
    rclpy.init(args=args)                            # ROS2 Python接口初始化
    node = ImageSubscriber("service_object_server")  # 创建ROS2节点对象并进行初始化
    rclpy.spin(node)                                 # 循环等待ROS2退出
    node.destroy_node()                              # 销毁节点对象
    rclpy.shutdown()                                 # 关闭ROS2 Python接口

完成代码的编写后需要设置功能包的编译选项，让系统知道Python程序的入口，打开功能包的setup.py文件，加入如下入口点的配置：
 

    entry_points={
        'console_scripts': [
         'service_adder_client  = learning_service.service_adder_client:main',
         'service_adder_server  = learning_service.service_adder_server:main',
         'service_object_client = learning_service.service_object_client:main',
         'service_object_server = learning_service.service_object_server:main',
        ],
    },

服务命令行操作

服务命令的常用操作如下：

$ ros2 service list                  # 查看服务列表
$ ros2 service type <service_name>   # 查看服务数据类型
$ ros2 service call <service_name> <service_type> <service_data>   # 发送服务请求

通信接口

在ROS系统中，无论是话题、服务还是后续要学习的动作机制，都涉及一个核心概念——通信接口。

通信本质上是多方之间的数据交互，而非单向传输。为了使交互过程更加高效规范，我们需要为传递的数据定义统一的结构标准，这就是通信接口的作用。

接口的定义

接口的概念在各个领域随处可见，无论是硬件结构还是软件开发，都有广泛的应用。

比如生活中最为常见的插头和插座，两者必须匹配才能使用，电脑和手机上的USB接口也是，什么Micro-USB、TypeC等等，都是关于接口的具体定义。

在软件开发中，接口的应用无处不在。例如，当我们编写程序时，函数及其输入输出参数就构成了典型的接口。每次调用函数，实际上就是通过这个接口将主程序与被调函数连接起来，使系统能够正常运行。

图形化编程中的模块化组件就是一个更直观的例子。每个模块都具有特定的结构和形状，只有当两个模块相互匹配时，才能实现协同工作。这种方式生动地展现了接口的概念。

因此，接口本质上是一种相互关系的体现。只有当双方彼此适配时，才能建立有效的连接。

ROS通信系统的核心目标是实现高效的数据传输，为此需要建立可靠的连接机制，并确保数据能够被准确封装和解析。基于这一需求，话题、服务等通信机制应运而生。这些机制传输的数据都必须遵循标准化的通信接口规范。

举例来说：

  • 摄像头驱动发布的图像话题由每个像素点的RGB三原色值组成
  • 控制机器人运动的速度指令包含线速度和角速度
  • 机器人配置服务则包括配置参数和反馈结果

ROS系统不仅提供了这些常用接口的定义，还支持用户自定义开发。这些标准化接口看似增加了约束，实则体现了ROS的设计精髓。以相机驱动节点为例，开发者无需了解底层实现细节，只需启动节点就能获取符合标准格式的图像数据，立即投入应用开发。同理，通过安装ROS键盘控制包，用户可直接获得标准化的速度控制指令，而无需关心其具体实现方式。

这种标准化设计极大地提升了开发效率，使开发者能够专注于核心功能的实现。

ROS通信接口

接口可以让程序之间的依赖降低，便于我们使用别人的代码，也方便别人使用我们的代码，这就是ROS的核心目标，减少重复造轮子。

ROS有三种常用的通信机制，分别是话题、服务、动作，通过每一种通信种定义的接口，各种节点才能有机的联系到一起。

语言无关

为了保证每一个节点可以使用不同语言编程，ROS将这些接口的设计做成了和语言无关的，比如这里看到的int32表示32位的整型数，int64表示64位的整型数，bool表示布尔值，还可以定义数组、结构体，这些定义在编译过程中，会自动生成对应到C++、Python等语言里的数据结构。

  • 话题通信接口的定义使用的是.msg文件，由于是单向传输，只需要描述传输的每一帧数据是什么就行，比如在这个定义里，会传输两个32位的整型数，x、y，我们可以用来传输二维坐标的数值。

  • 服务通信接口的定义使用的是.srv文件，包含请求和应答两部分定义，通过中间的“---”区分，比如之前我们学习的加法求和功能，请求数据是两个64位整型数a和b，应答是求和的结果sum。

  • 动作是另外一种通信机制，用来描述机器人的一个运动过程，使用.action文件定义，比如我们让小海龟转90度，一边转一边周期反馈当前的状态，此时接口的定义分成了三个部分，分别是动作的目标，比如是开始运动，运动的结果，最终旋转的90度是否完成，还有一个周期反馈，比如每隔1s反馈一下当前转到第10度、20度还是30度了，让我们知道运动的进度。

标准接口

大家可能好奇ROS系统到底给我们定义了哪些接口呢？我们可以在ROS安装路径中的share文件夹中找到，涵盖众多标准定义，大家可以打开几个看看。

案例一：服务接口的定义和使用

了解了通信接口的概念，接下来我们再从代码实现的角度，研究下如何定义以及使用一个接口。

在之前服务概念讲解的课程中，我们编写了这样一个例程，我们再来回顾下。

有三个节点，第一个驱动相机发布图像话题，第二个是机器视觉识别节点，封装了一个服务的服务端对象，提供目标识别位置的查询服务，第三个节点在需要目标位置的时候，就可以发送请求，收到位置进行使用了。

接口定义

在这个例程中，我们使用GetObjectPosition.srv定义了服务通信的接口：

learning_interface/srv/GetObjectPosition.srv

bool get      # 获取目标位置的指令
---
int32 x       # 目标的X坐标
int32 y       # 目标的Y坐标

定义中有两个部分，上边是获取目标位置的指令，get为true的话，就表示我们需要一次位置，服务端就会反馈这个x、y坐标了。

完成定义后，还需要在功能包的CMakeLists.txt中配置编译选项，让编译器在编译过程中，根据接口定义，自动生成不同语言的代码：

...

find_package(rosidl_default_generators REQUIRED)

rosidl_generate_interfaces(${PROJECT_NAME}
  "srv/GetObjectPosition.srv"
)

...

功能包的package.xml文件中也需要添加代码生成的功能依赖：


 <build_depend>rosidl_default_generators</build_depend>
 <exec_depend>rosidl_default_runtime</exec_depend>
 <member_of_group>rosidl_interface_packages</member_of_group>

程序调用

我们在代码中再来重点看下接口的使用方法。

客户端接口调用

learning_service/service_object_client.py

import rclpy                                            # ROS2 Python接口库
from rclpy.node   import Node                           # ROS2 节点类
from learning_interface.srv import GetObjectPosition    # 自定义的服务接口

class objectClient(Node):
    def __init__(self, name):
        super().__init__(name)                          # ROS2节点父类初始化
        self.client = self.create_client(GetObjectPosition, 'get_target_position')
        while not self.client.wait_for_service(timeout_sec=1.0):
            self.get_logger().info('service not available, waiting again...')
        self.request = GetObjectPosition.Request()

    def send_request(self):
        self.request.get = True
        self.future = self.client.call_async(self.request)

def main(args=None):
    rclpy.init(args=args)                             # ROS2 Python接口初始化
    node = objectClient("service_object_client")      # 创建ROS2节点对象并进行初始化
    node.send_request()

    while rclpy.ok():
        rclpy.spin_once(node)

        if node.future.done():
            try:
                response = node.future.result()
            except Exception as e:
                node.get_logger().info(
                    'Service call failed %r' % (e,))
            else:
                node.get_logger().info(
                    'Result of object position:\n x: %d y: %d' %
                    (response.x, response.y))
            break
    node.destroy_node()                              # 销毁节点对象
    rclpy.shutdown()                                 # 关闭ROS2 Python接口
服务端接口调用

learning_service/service_object_server.py

import rclpy                                           # ROS2 Python接口库
from rclpy.node import Node                            # ROS2 节点类
from sensor_msgs.msg import Image                      # 图像消息类型
import numpy as np                                     # Python数值计算库
from cv_bridge import CvBridge                         # ROS与OpenCV图像转换类
import cv2                                             # Opencv图像处理库
from learning_interface.srv import GetObjectPosition   # 自定义的服务接口

lower_red = np.array([0, 90, 128])     # 红色的HSV阈值下限
upper_red = np.array([180, 255, 255])  # 红色的HSV阈值上限

class ImageSubscriber(Node):
    def __init__(self, name):
        super().__init__(name)                              # ROS2节点父类初始化
        self.sub = self.create_subscription(
            Image, 'image_raw', self.listener_callback, 10) # 创建订阅者对象（消息类型、话题名、订阅者回调函数、队列长度）
        self.cv_bridge = CvBridge()                         # 创建一个图像转换对象，用于OpenCV图像与ROS的图像消息的互相转换

        self.srv = self.create_service(GetObjectPosition,   # 创建服务器对象（接口类型、服务名、服务器回调函数）
                                       'get_target_position',
                                       self.object_position_callback)    
        self.objectX = 0
        self.objectY = 0                              

    def object_detect(self, image):
        hsv_img = cv2.cvtColor(image, cv2.COLOR_BGR2HSV)      # 图像从BGR颜色模型转换为HSV模型
        mask_red = cv2.inRange(hsv_img, lower_red, upper_red) # 图像二值化
        contours, hierarchy = cv2.findContours(
            mask_red, cv2.RETR_LIST, cv2.CHAIN_APPROX_NONE)   # 图像中轮廓检测

        for cnt in contours:                                  # 去除一些轮廓面积太小的噪声
            if cnt.shape[0] < 150:
                continue

            (x, y, w, h) = cv2.boundingRect(cnt)              # 得到苹果所在轮廓的左上角xy像素坐标及轮廓范围的宽和高
            cv2.drawContours(image, [cnt], -1, (0, 255, 0), 2)# 将苹果的轮廓勾勒出来
            cv2.circle(image, (int(x+w/2), int(y+h/2)), 5,
                       (0, 255, 0), -1)                       # 将苹果的图像中心点画出来

            self.objectX = int(x+w/2)
            self.objectY = int(y+h/2)

        cv2.imshow("object", image)                            # 使用OpenCV显示处理后的图像效果
        cv2.waitKey(50)

    def listener_callback(self, data):
        self.get_logger().info('Receiving video frame')        # 输出日志信息，提示已进入回调函数
        image = self.cv_bridge.imgmsg_to_cv2(data, 'bgr8')     # 将ROS的图像消息转化成OpenCV图像
        self.object_detect(image)                              # 苹果检测

    def object_position_callback(self, request, response):     # 创建回调函数，执行收到请求后对数据的处理
        if request.get == True:
            response.x = self.objectX                          # 目标物体的XY坐标
            response.y = self.objectY
            self.get_logger().info('Object position\nx: %d y: %d' %
                                   (response.x, response.y))   # 输出日志信息，提示已经反馈
        else:
            response.x = 0
            response.y = 0
            self.get_logger().info('Invalid command')          # 输出日志信息，提示已经反馈
        return response


def main(args=None):                                 # ROS2节点主入口main函数
    rclpy.init(args=args)                            # ROS2 Python接口初始化
    node = ImageSubscriber("service_object_server")  # 创建ROS2节点对象并进行初始化
    rclpy.spin(node)                                 # 循环等待ROS2退出
    node.destroy_node()                              # 销毁节点对象
    rclpy.shutdown()                                 # 关闭ROS2 Python接口

案例二：话题接口的定义与使用

话题通信接口的定义也是类似的，继续从之前的机器视觉案例中来衍生，我们想把服务换成话题，周期发布目标识别的位置，不管有没有人需要。

运行效果

现在我们会运行三个节点：

  • 第一个节点，将驱动相机并发布图像话题，此时的话题数据使用的是ROS中标准定义的Image图像消息；
  • 第二个节点，会运行视觉识别功能，识别目标的位置，这个位置我们希望封装成话题消息，发布出去，谁需要使用谁就来订阅；
  • 第三个节点，订阅位置话题，打印到终端中。

启动三个终端，分别运行以上节点：

$ ros2 run usb_cam usb_cam_node_exe
$ ros2 run learning_topic interface_object_pub
$ ros2 run learning_topic interface_object_sub

接口定义

在这个例程中，我们使用ObjectPosition.msg定义了服务通信的接口：

learning_interface/msg/ObjectPosition.msg

int32 x      # 表示目标的X坐标
int32 y      # 表示目标的Y坐标

话题消息的内容是一个位置，我们使用x、y坐标值进行描述。

完成定义后，还需要在功能包的CMakeLists.txt中配置编译选项，让编译器在编译过程中，根据接口定义，自动生成不同语言的代码：

...

find_package(rosidl_default_generators REQUIRED)

rosidl_generate_interfaces(${PROJECT_NAME}
  "msg/ObjectPosition.msg"
)

...

程序调用

我们在代码中再来重点看下接口的使用方法。

发布者接口调用

learning_topic/interface_object_pub.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
@作者: 古月居(www.guyuehome.com)
@说明: ROS2接口示例-发布目标位置
"""

import rclpy                                       # ROS2 Python接口库
from rclpy.node import Node                        # ROS2 节点类
from sensor_msgs.msg import Image                  # 图像消息类型
from cv_bridge import CvBridge                     # ROS与OpenCV图像转换类
import cv2                                         # Opencv图像处理库
import numpy as np                                 # Python数值计算库
from learning_interface.msg import ObjectPosition  # 自定义的目标位置消息

lower_red = np.array([0, 90, 128])                 # 红色的HSV阈值下限
upper_red = np.array([180, 255, 255])              # 红色的HSV阈值上限

"""
创建一个订阅者节点
"""
class ImageSubscriber(Node):

    def __init__(self, name):
        super().__init__(name)                                  # ROS2节点父类初始化
        self.sub = self.create_subscription(
            Image, 'image_raw', self.listener_callback, 10)     # 创建订阅者对象（消息类型、话题名、订阅者回调函数、队列长度）
        self.pub = self.create_publisher(
            ObjectPosition, "object_position", 10)              # 创建发布者对象（消息类型、话题名、队列长度）
        self.cv_bridge = CvBridge()                             # 创建一个图像转换对象，用于OpenCV图像与ROS的图像消息的互相转换

        self.objectX = 0
        self.objectY = 0   

    def object_detect(self, image):      
        hsv_img = cv2.cvtColor(image, cv2.COLOR_BGR2HSV)        # 图像从BGR颜色模型转换为HSV模型
        mask_red = cv2.inRange(hsv_img, lower_red, upper_red)   # 图像二值化
        contours, hierarchy = cv2.findContours(
            mask_red, cv2.RETR_LIST, cv2.CHAIN_APPROX_NONE)     # 图像中轮廓检测
        for cnt in contours:                                    # 去除一些轮廓面积太小的噪声
            if cnt.shape[0] < 150:
                continue

            (x, y, w, h) = cv2.boundingRect(cnt)                # 得到苹果所在轮廓的左上角xy像素坐标及轮廓范围的宽和高

            cv2.drawContours(image, [cnt], -1, (0, 255, 0), 2)  # 将苹果的轮廓勾勒出来
            cv2.circle(image, (int(x+w/2), int(y+h/2)), 5,      # 将苹果的图像中心点画出来
                       (0, 255, 0), -1)   

            self.objectX = int(x+w/2)
            self.objectY = int(y+h/2)

        cv2.imshow("object", image)                             # 使用OpenCV显示处理后的图像效果
        cv2.waitKey(50)

    def listener_callback(self, data):
        self.get_logger().info('Receiving video frame')         # 输出日志信息，提示已进入回调函数
        image = self.cv_bridge.imgmsg_to_cv2(data, 'bgr8')      # 将ROS的图像消息转化成OpenCV图像
        position = ObjectPosition()
        self.object_detect(image)                               # 苹果检测
        position.x, position.y = int(self.objectX), int(self.objectY)
        self.pub.publish(position)                              # 发布目标位置

def main(args=None):                                        # ROS2节点主入口main函数
    rclpy.init(args=args)                                   # ROS2 Python接口初始化
    node = ImageSubscriber("topic_webcam_sub")              # 创建ROS2节点对象并进行初始化
    rclpy.spin(node)                                        # 循环等待ROS2退出
    node.destroy_node()                                     # 销毁节点对象
    rclpy.shutdown()                                        # 关闭ROS2 Python接口
订阅者接口调用

learning_topic/interface_object_sub.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
@作者: 古月居(www.guyuehome.com)
@说明: ROS2接口示例-订阅目标位置
"""

import rclpy                                       # ROS2 Python接口库
from rclpy.node   import Node                      # ROS2 节点类
from std_msgs.msg import String                    # 字符串消息类型
from learning_interface.msg import ObjectPosition  # 自定义的目标位置消息

"""
创建一个订阅者节点
"""
class SubscriberNode(Node):
    def __init__(self, name):
        super().__init__(name)                                                # ROS2节点父类初始化
        self.sub = self.create_subscription(\
            ObjectPosition, "/object_position", self.listener_callback, 10)   # 创建订阅者对象（消息类型、话题名、订阅者回调函数、队列长度

    def listener_callback(self, msg):                                         # 创建回调函数，执行收到话题消息后对数据的处理
        self.get_logger().info('Target Position: "(%d, %d)"' % (msg.x, msg.y))# 输出日志信息，提示订阅收到的话题消息


def main(args=None):                                 # ROS2节点主入口main函数
    rclpy.init(args=args)                            # ROS2 Python接口初始化
    node = SubscriberNode("interface_position_sub")  # 创建ROS2节点对象并进行初始化
    rclpy.spin(node)                                 # 循环等待ROS2退出
    node.destroy_node()                              # 销毁节点对象
    rclpy.shutdown()                                 # 关闭ROS2 Python接口

接口命令行操作

接口命令的常用操作如下：

$ ros2 interface list                    # 查看系统接口列表
$ ros2 interface show <interface_name>   # 查看某个接口的详细定义
$ ros2 interface package <package_name>  # 查看某个功能包中的接口定义

image-20220528004610983

image-20220528004525471

image-20220528004538865

# python # 开发语言
Logo
DAMO开发者矩阵

DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐

2026最新8款团队编程效率提升工具：基础版免费深度实测合集

今年8月一个周一早上9点20分，我带的6人研发团队要在午前修复代码审查机器人、补齐测试并更新接口文档。过去这类多文件修改经常要占掉半天，我便把同一任务交给8款工具实测。据官方产品信息，字节跳动出品的 TRAE 基础版免费，是国内首款 AI 原生 IDE，正好适合验证团队编程效率提升工具能否真正缩短交付链路。

avatar DAMO开发者矩阵
cover

边缘计算物理抽象：基于GPIO隔离实现非侵入式机器人梯控

avatar DAMO开发者矩阵
cover

企业微信 AI 行为与意图监测怎么做？员工行为检测 + 客户意图识别 + 服务质检三合一

avatar DAMO开发者矩阵