警告

您正在阅读的 ROS 2 文档版本已达到 EOL（生命周期结束），不再受官方支持。如果您想了解最新信息，请访问 Jazzy.

高效的进程内通信

背景介绍

ROS 应用程序通常由单个 "节点 "组成，这些 "节点 "只能执行有限的任务，并与系统的其他部分解耦。这有利于故障隔离、更快的开发、模块化和代码重用，但往往要以牺牲性能为代价。在 ROS 1 初步开发完成后，高效节点组合的需求变得非常明显，于是 Nodelets 应运而生。在 ROS 2 中，我们旨在通过解决一些需要重组节点的基本问题来改进 Nodelets 的设计。

在本演示中，我们将重点介绍如何通过单独定义节点，并在不更改节点代码或限制其功能的情况下，将其组合到不同的流程布局中，从而手动组成节点。

制作演示

这些演示应该可以在三大操作系统（Windows、Mac 或 Linux）中的任何一个上运行。其中有些演示需要安装 OpenCV。

使用预制二进制文件

如果您已经安装了二进制文件，只需从 ROS 2 安装文件中获取源代码，然后跳到任何一个单独的演示中查看如何运行它们。

从源头开始建设

确保已安装 OpenCV，然后按照源代码说明进行操作。您可以从主页面的 ros2 安装页面.构建完成后，您可以调用设置文件，然后继续往下查看具体的演示程序，了解它们的相关信息，并获取运行说明。

运行和理解演示

有几个不同的演示：有些是玩具问题，旨在突出进程内通信功能的特点；有些是端到端的示例，使用 OpenCV 并演示将节点重新组合为不同配置的能力。

双节点管道演示

该演示旨在展示进程内的发布/订阅连接在使用 std::unique_ptrs.

首先让我们来看看消息来源：

https://github.com/ros2/demos/blob/dashing/intra_process_demo/src/two_node_pipeline/two_node_pipeline.cpp

#include 时间顺序<chrono>；
#include <cinttypes>；
#include <cstdio>；
#include <内存>；
#include <字符串>；
#include <utility>；

#include "rclcpp/rclcpp.hpp"；
#include "std_msgs/msg/int32.hpp"；

使用 命名空间 标准::计时器;

// 生成信息的节点。
结构 制片人 : 公 rclcpp::节点
{
  制片人(缢 标准::字符串 及样品； 名字, 缢 标准::字符串 及样品； 产量)
  : 节点(名字, rclcpp::节点选项().使用内部流程通信(真))
  {
    // 在输出主题上创建一个发布者。
    公共_ = 此->；创建出版商<；std_msgs::信息::Int32>；(产量, 10);
    标准::弱_ptr<；标准::删除指针<；decltype(公共_.获取())>:：类型>； 捕获的出版物 = 公共_;
    // 创建一个定时器，以 ~1Hz 的频率在输出主题上发布。
    汽车 回调 = [捕获的出版物]() ->； 空白 {
        汽车 pub_ptr = 捕获的出版物.水闸();
        如果 (!pub_ptr) {
          返回;
        }
        天电 int32_t 计数 = 0;
        std_msgs::信息::Int32::UniquePtr 信息(新 std_msgs::信息::Int32());
        信息->；数据 = 计数++;
        printf(
          "Published message with value: %d, and address：0x%"； PRIXPTR "；\n"；, 信息->；数据,
          重读<；标准::uintptr_t>；(信息.获取()));
        pub_ptr->；发布(标准::行动(信息));
      };
    定时器 = 此->；创建隔离墙计时器(1s, 回调);
  }

  rclcpp::出版商<；std_msgs::信息::Int32>:：SharedPtr 公共_;
  rclcpp::定时器基数::SharedPtr 定时器;
};

// 消耗信息的节点。
结构 消费者 : 公 rclcpp::节点
{
  消费者(缢 标准::字符串 及样品； 名字, 缢 标准::字符串 及样品； 输入)
  : 节点(名字, rclcpp::节点选项().使用内部流程通信(真))
  {
    // 在输入主题上创建订阅，在收到新信息时打印。
    子_ = 此->；创建订阅<；std_msgs::信息::Int32>；(
      输入,
      10,
      [](std_msgs::信息::Int32::UniquePtr 信息) {
        printf(
          "收到的信息值为：%d，地址为：0x%"：0x%"； PRIXPTR "；\n"；, 信息->；数据,
          重读<；标准::uintptr_t>；(信息.获取()));
      });
  }

  rclcpp::订阅<；std_msgs::信息::Int32>:：SharedPtr 子_;
};

int 主要(int 参数, 烧焦 * 参数[])
{
  setvbuf(数据输出, NULL, _IONBF, BUFSIZ);
  rclcpp::启动(参数, 参数);
  rclcpp::执行人::单线程执行器 执行人;

  汽车 监制 = 标准::共享<；制片人>；("制作人"；, "号码"；);
  汽车 消费者 = 标准::共享<；消费者>；("消费者"；, "号码"；);

  执行人.添加节点(监制);
  执行人.添加节点(消费者);
  执行人.后旋();

  rclcpp::关闭();

  返回 0;
}

通过查看 主要 函数，我们有一个生产者节点和一个消费者节点，我们将它们添加到一个单线程执行器中，然后调用 spin。

如果你查看 "生产者 "节点在 制片人 在结构图中，您可以看到我们创建了一个在 "数字 "主题上发布信息的发布器和一个定时器，定时器会定期创建一条新信息，打印出其在内存中的地址和内容值，然后将其发布。

消费者 "节点比较简单，你可以在 消费者 结构，因为它只订阅 "数字 "主题，并打印收到的信息的地址和值。

我们期望生产者打印出一个地址和值，消费者打印出一个匹配的地址和值。这表明进程内通信确实有效，至少对于简单的图形来说，避免了不必要的复制。

让我们运行演示，执行 玫瑰2 运行 进程内演示 双节点管道 可执行文件（别忘了先获取设置文件的源代码）：

$ 玫瑰2 运行 进程内演示 双节点管道 已发布 信息 与 价值： 0, 和 地址 0x7fb02303faf0 已发布 信息 与 价值： 1, 和 地址 0x7fb020cf0520
 已收到 信息 与 价值： 1, 和 地址 0x7fb020cf0520 已发布 信息 与 价值： 2, 和 地址 0x7fb020e12900
 已收到 信息 与 价值： 2, 和 地址 0x7fb020e12900 已发布 信息 与 价值： 3, 和 地址 0x7fb020cf0520
 已收到 信息 与 价值： 3, 和 地址 0x7fb020cf0520 已发布 信息 与 价值： 4, 和 地址 0x7fb020e12900
 已收到 信息 与 价值： 4, 和 地址 0x7fb020e12900 已发布 信息 与 价值： 5, 和 地址 0x7fb02303cea0
 已收到 信息 与 价值： 5, 和 地址 0x7fb02303cea0
[...]

你会注意到，这些信息大约每秒一条。这是因为我们告诉定时器大约每秒触发一次。

您可能还注意到，第一条信息（值为 0)没有相应的 "收到消息...... "行。这是因为发布/订阅是 "尽力而为 "的，我们没有启用任何类似 "锁定 "的行为。这意味着，如果发布者在订阅建立之前发布了一条消息，订阅将不会收到该消息。这种竞赛条件会导致前几条信息丢失。在这种情况下，由于信息每秒只发送一次，通常只会丢失第一条信息。

最后，您可以看到 "已发布邮件...... "和 "已接收邮件...... "两行的值相同，地址也相同。这表明接收到的信息的地址与已发布的信息的地址相同，而不是副本。这是因为我们是用 std::unique_ptr可以安全地在系统中移动消息的所有权。您还可以使用 缢 及样品； 和 std::shared_ptr但在这种情况下不会出现零拷贝。

循环流水线演示

这个演示与上一个演示类似，但生产者不是为每次迭代创建新消息，而是只使用一个消息实例。这是通过在图中创建一个循环，并在旋转执行器之前从外部让其中一个节点发布消息，从而 "启动 "通信：

https://github.com/ros2/demos/blob/dashing/intra_process_demo/src/cyclic_pipeline/cyclic_pipeline.cpp

#include 时间顺序<chrono>；
#include <cinttypes>；
#include <cstdio>；
#include <内存>；
#include <字符串>；
#include <utility>；

#include "rclcpp/rclcpp.hpp"；
#include "std_msgs/msg/int32.hpp"；

使用 命名空间 标准::计时器;

// 该节点接收一个 Int32，等待 1 秒，然后递增并发送。
结构 递增管道 : 公 rclcpp::节点
{
  递增管道(缢 标准::字符串 及样品； 名字, 缢 标准::字符串 及样品； 于, 缢 标准::字符串 及样品； 向外)
  : 节点(名字, rclcpp::节点选项().使用内部流程通信(真))
  {
    // 在输出主题上创建一个发布者。
    酒吧 = 此->；创建出版商<；std_msgs::信息::Int32>；(向外, 10);
    标准::弱_ptr<；标准::删除指针<；decltype(酒吧.获取())>:：类型>； 捕获的出版物 = 酒吧;
    // 在输入主题上创建订阅。
    字幕 = 此->；创建订阅<；std_msgs::信息::Int32>；(
      于,
      10,
      [捕获的出版物](std_msgs::信息::Int32::UniquePtr 信息) {
        汽车 pub_ptr = 捕获的出版物.水闸();
        如果 (!pub_ptr) {
          返回;
        }
        printf(
          "收到信息，信息值为：%d，地址为：0x%"：0x%"； PRIXPTR "；\n"；, 信息->；数据,
          重读<；标准::uintptr_t>；(信息.获取()));
        printf(睡眠 1 秒...\n"；);
        如果 (!rclcpp::sleep_for(1s)) {
          返回;    // 如果睡眠失败（例如按 ctrl-c 键），则返回。
        }
        printf(完成。\n"；);
        信息->；数据++;    // 增加信息数据。
        printf(
          "递增并以值：%d 和地址发送：0x%"； PRIXPTR "；\n"；, 信息->；数据,
          重读<；标准::uintptr_t>；(信息.获取()));
        pub_ptr->；发布(标准::行动(信息));    // 将信息发送到输出主题。
      });
  }

  rclcpp::出版商<；std_msgs::信息::Int32>:：SharedPtr 酒吧;
  rclcpp::订阅<；std_msgs::信息::Int32>:：SharedPtr 字幕;
};

int 主要(int 参数, 烧焦 * 参数[])
{
  setvbuf(数据输出, NULL, _IONBF, BUFSIZ);
  rclcpp::启动(参数, 参数);
  rclcpp::执行人::单线程执行器 执行人;

  // 通过连接两个 IncrementerPipe's 的输入和输出主题，创建一个简单的循环。
  // 期望它们之间传递的信息地址永远不会改变。
  汽车 管道1 = 标准::共享<；递增管道>；("pipe1"；, "topic1"；, "topic2"；);
  汽车 管道2 = 标准::共享<；递增管道>；("pipe2"；, "topic2"；, "topic1"；);
  rclcpp::sleep_for(1s);  // 等待订阅建立，以避免出现竞赛条件。
  // 发布第一条信息（启动循环）。
  标准::唯一参数<；std_msgs::信息::Int32>； 信息(新 std_msgs::信息::Int32());
  信息->；数据 = 42;
  printf(
    "Published first message with value: %d, and address：0x%"； PRIXPTR "；\n"；, 信息->；数据,
    重读<；标准::uintptr_t>；(信息.获取()));
  管道1->；酒吧->；发布(标准::行动(信息));

  执行人.添加节点(管道1);
  执行人.添加节点(管道2);
  执行人.后旋();

  rclcpp::关闭();

  返回 0;
}

与上一个演示不同，本演示只使用了一个节点，并以不同的名称和配置实例化了两次。图最终是 管道1 ->； 管道2 ->； 管道1 ......循环往复。

线路 pipe1->pub->publish(msg)； 启动进程，但从那时起，信息会通过每个节点在自己的订阅回调中调用发布在节点之间来回传递。

这里的预期是节点每秒来回传递一次信息，每次都会递增信息的值。因为消息是以 唯一参数 在开始时创建的同一信息会被持续使用。

为了验证这些期望，让我们运行它：

% 玫瑰2 运行 进程内演示 cyclic_pipeline 已发布 第一次 信息 与 价值：  42, 和 地址 0x7fd2ce0a2bc0 收到 信息 与 价值：         42, 和 地址 0x7fd2ce0a2bc0
  睡觉 对于 1 第二...
  完成的.递增 和 发送 与 价值： 43, 和 地址 0x7fd2ce0a2bc0 收到 信息 与 价值：         43, 和 地址 0x7fd2ce0a2bc0
  睡觉 对于 1 第二...
  完成的.递增 和 发送 与 价值： 44, 和 地址 0x7fd2ce0a2bc0 收到 信息 与 价值：         44, 和 地址 0x7fd2ce0a2bc0
  睡觉 对于 1 第二...
  完成的.递增 和 发送 与 价值： 45, 和 地址 0x7fd2ce0a2bc0 收到 信息 与 价值：         45, 和 地址 0x7fd2ce0a2bc0
  睡觉 对于 1 第二...
  完成的.递增 和 发送 与 价值： 46, 和 地址 0x7fd2ce0a2bc0 收到 信息 与 价值：         46, 和 地址 0x7fd2ce0a2bc0
  睡觉 对于 1 第二...
  完成的.递增 和 发送 与 价值： 47, 和 地址 0x7fd2ce0a2bc0 收到 信息 与 价值：         47, 和 地址 0x7fd2ce0a2bc0
  睡觉 对于 1 第二...
[...]

每次迭代时，你都会看到数字不断增加，从 42 开始......因为是 42，而且一直重复使用相同的信息，指针地址不会改变，这就避免了不必要的复制。

图像管道演示

在本演示中，我们将使用 OpenCV 捕捉、注释和查看图像。

macOS 用户注意：如果这些示例不起作用或出现类似以下的错误 ddsi_conn_write 失败 -1 则需要增加系统范围内的 UDP 数据包大小：

$ 苏都 sysctl -w net.inet.udp.recvspace=209715
$ 苏都 sysctl -w net.inet.udp.maxdgram=65500

重新启动后，这些更改将不会持续。

简单管道

首先，我们将有一个由三个节点组成的管道，排列如下： 摄像机节点 ->； 水印节点 ->； 图像视图节点

"(《世界人权宣言》) 摄像机节点 从摄像设备读取数据 0 在计算机上写入图像信息并发布。图像 水印节点 的输出。 摄像机节点 并在发布前添加更多文字。最后 图像视图节点 的输出。 水印节点在图像中写入更多文字，然后用 cv::imshow.

在每个节点中，正在发送或已经接收或两者兼有的信息地址被写入图像。水印节点和图像视图节点的设计目的是在不复制图像的情况下修改图像，因此，只要节点处于同一流程中，图像上印记的地址都应该是相同的，而且图的组织结构也应保持如上所示的流水线。

备注

在某些系统上（我们在 Linux 系统上看到过这种情况），打印到屏幕上的地址可能不会改变。这是因为重复使用了同一个唯一指针。在这种情况下，管道仍在运行。

让我们执行以下可执行文件来运行演示：

玫瑰2 运行 进程内演示 统一图像管道

你应该看到这样的内容：

按空格键可以暂停图像渲染，再次按空格键可以继续渲染。您还可以按 q 或 ESC 退出。

如果暂停图像查看器，就可以比较图像上写的地址，发现它们是一样的。

带有两个图像查看器的管道

现在我们来看一个与上面类似的示例，只不过它有两个图像视图节点。所有节点仍在同一进程中，但现在应显示两个图像视图窗口。(macOS用户请注意：图像视图窗口可能会相互重叠）。让我们使用命令运行它：

玫瑰2 运行 进程内演示 带有两个图像视图的图像管道

与上一个示例一样，您可以按空格键暂停渲染，然后再按一次空格键继续渲染。您可以停止更新，检查写入屏幕的指针。

正如您在上面的示例图片中看到的那样，我们有一张所有指针都相同的图片，而另一张图片的前两个条目的指针与第一张图片相同，但第二张图片的最后一个指针却不同。要理解为什么会出现这种情况，请考虑一下图的拓扑结构：

摄像机节点 ->； 水印节点 ->； 图像视图节点
                              ->； 图像视图节点 2

之间的联系 摄像机节点 和 水印节点 可以使用相同的指针而无需复制，因为只有一个进程内订阅可以接收到报文。但对于 水印节点 和两个图像视图节点的关系是一对多，因此如果图像视图节点使用 唯一参数 那么就不可能将同一指针的所有权同时传递给两个回调。不过，它可以传递给其中一个。至于哪一个会获得原始指针，并没有定义，而只是最后一个被传递的指针。

请注意，图像视图节点没有订阅 唯一参数 回调。而不是用 缢 共享_ptrs.这意味着系统提供的 共享_ptr 这两个回调。处理第一个进程内订阅时，内部存储的 唯一参数 晋升为 共享_ptr.每个回调都将接收同一消息的共享所有权。

带有进程间查看器的管道

还有一件重要的事情要做好，那就是避免在进程间订阅时中断进程内的零拷贝行为。为了测试这一点，我们可以运行第一个图像管道演示、 统一图像管道然后运行一个独立的 图像视图节点 (别忘了在前缀中加上 玫瑰2 运行 进程内演示 在终端中）。看起来会是这样

很难同时暂停两幅图像，因此图像可能不一致，但需要注意的是 统一图像管道 图像视图显示每个步骤的相同地址。这意味着，即使也订阅了外部视图，进程内的零拷贝也会保留。您还可以看到，进程间图像视图的前两行文本和第三行文本中独立图像查看器的进程 ID 不同。

展望未来

这些演示为我们正在积极开发的一些很酷的新功能奠定了基础，但现在还缺少一些东西。

改进空间

让我们先来看看 OSRF 知道我们可以做得更好或有什么不同，然后再继续往下看。

性能、性能、性能

这是一份非常粗糙的初稿。除了上面列举的内容，还有很多需要改进的地方。随着我们深入研究系统细节，更好地了解中间件供应商的具体做法，并尝试实施内部流程的其他策略，我们将开始提高性能。

闭锁

我们还没有完全实现闩锁的概念，但我们很可能需要调整流程内管理器的实现，以考虑到流程内延迟订阅也应交付的事实。关于如何做到这一点，有几种选择，我们将在不久的将来进行一些测试并找出解决方案。

超越酒吧/分会

在服务、参数或操作方面，我们还没有做任何这方面的工作，但我们会做的。

伪装类型

这是即将推出的最酷的功能之一，我们在演示中没有介绍到。想象一下上面的图像管道演示，但不是通过 传感器_msgs/图像的周围，你正在发布和订阅 cv::Mat 对象。这在 ROS 1 中已经存在，请参见： https://wiki.ros.org/roscpp/Overview/MessagesSerializationAndAdaptingTypes

在 ROS 1 中，这是通过在处理第三方类型时对其进行序列化/反序列化来实现的。这就意味着，在节点程序之间传递过程中，你需要对intra进程进行序列化。但在 ROS 2 中，我们希望尽可能以最高效的方式实现这一点。就像这些演示中展示的那样，在某些情况下，消息的实例可以在整个管道中使用，我们也希望在第三方类型中做到这一点。因此，我们可以设想在图像管道中使用单个 cv::Mat 中间件永远不会复制它。要做到这一点，需要在内部进程管理器中增加一些智能，但我们已经有了一个设计方案，并正在进行一些概念验证。

有了这些功能，希望有一天您可以相信中间件会尽可能高效地处理您的数据。这样，您就可以在不牺牲模块性或自省性的前提下编写高性能算法！