跨进程通信使用 Zenoh中间件 进行高效数据传输的测试和分析

1. 引言

Zenoh 是一个高效的数据传输框架，核心由 Rust 编写，同时提供了 C++、Python 等多种语言绑定。支持多种部署环境和跨网络通信，实现了去中心化的自适应网络发现，最小的协议网络开销，能够满足低延迟和高吞吐量的要求，甚至它内部也实现了IPC(zero copy)。
本文将探讨如何在 C++ 环境下使用 Zenoh 进行数据传输，并讨论 Zenoh 与 ROS2 的集成方式。

2. Zenoh C++ 使用指南

2.1 安装 Zenoh C++ 库

首先，获取并安装 Zenoh C++ 库：

# 获取 zenoh-cpp 库
git clone https://github.com/eclipse-zenoh/zenoh-cpp.git
cd zenoh-cpp# 安装依赖并构建
mkdir build && cd build
cmake ..
make
sudo make install

2.2 编写基本的 Zenoh C++ 程序

一个基本的 Zenoh C++ 程序包括以下步骤：

1. 初始化 Zenoh 会话。
2. 发布或订阅数据。
3. 处理数据。

订阅示例

#include <iostream>
#include <zenoh/zenoh.hpp>int main(int argc, char *argv[]) {// 初始化 Zenoh sessionauto config = zenoh::Config();auto z = zenoh::open(config).res();// 订阅数据auto sub = z->subscribe("/example/key", [](const zenoh::Sample& sample) {std::cout << "Received data: " << sample.payload.data() << std::endl;});// 等待数据std::cout << "Waiting for data..." << std::endl;std::this_thread::sleep_for(std::chrono::minutes(10)); // 保持运行return 0;
}

发布示例

#include <iostream>
#include <zenoh/zenoh.hpp>int main(int argc, char *argv[]) {// 初始化 Zenoh sessionauto config = zenoh::Config();auto z = zenoh::open(config).res();// 发布数据for (int i = 0; i < 10; ++i) {std::string message = "Hello Zenoh: " + std::to_string(i);z->put("/example/key", message);std::cout << "Published: " << message << std::endl;std::this_thread::sleep_for(std::chrono::seconds(1));}return 0;
}

2.3 编译和运行程序

编译时需要链接 Zenoh 库：

g++ -std=c++17 -o zenoh_example zenoh_example.cpp -lzenoh

然后运行程序：

./zenoh_example

3. Zenoh 与 ROS2 集成

Zenoh 提供了一个轻量级且高效的消息传输机制，能够满足低延迟、高吞吐量的要求。通过 Zenoh 作为 ROS2 的 RMW (ROS Middleware) 实现，开发者可以实现跨网络的高效通信。Zenoh 的架构分为核心通信引擎和不同语言的绑定接口，其中核心由 Rust 编写，C++ 和 Python 绑定则为主流语言提供支持。

在 ROS2 中，使用 Zenoh 的典型步骤包括：

1. 安装 Zenoh 和 ROS2 的 Zenoh RMW 实现。
2. 配置 ROS2 使用 Zenoh 作为默认中间件。
3. 编写基于 Zenoh 的 ROS2 应用程序。

要实现 Zenoh 与 ROS2 的集成，可以按照以下步骤进行详细配置和开发：

3.1 安装 Zenoh

Zenoh 的核心库可以通过以下命令安装：

git clone https://github.com/eclipse-zenoh/zenoh.git
cd zenoh
cargo build --release

Zenoh 的核心是用 Rust 编写的，因此需要安装 Rust 和 Cargo。如果还未安装，可以通过以下命令安装：

curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh

3.2 安装 ROS2 的 Zenoh RMW 实现

ROS2 默认支持多种 RMW（ROS Middleware）实现，Zenoh 也可以作为其中之一使用。你需要安装 rmw_zenoh 和 rmw_zenoh_cpp：

git clone https://github.com/eclipse-zenoh/zenoh-plugin-dds.git
cd zenoh-plugin-dds
cargo build --release

然后，可以将编译好的库添加到 ROS2 的路径中。

3.3 设置 RMW 实现为 Zenoh

在运行 ROS2 节点时，可以通过以下命令将 RMW_IMPLEMENTATION 设置为 rmw_zenoh_cpp：

export RMW_IMPLEMENTATION=rmw_zenoh_cpp

也可以在 .bashrc 文件中添加此命令，以便每次启动终端时自动配置。

3.4 验证配置

通过以下命令验证 Zenoh 是否已作为 ROS2 的默认中间件：

ros2 doctor --report

输出中应显示 RMW_IMPLEMENTATION: rmw_zenoh_cpp。

4. 编写基于 Zenoh 的 ROS2 应用程序

在 ROS2 中，你可以按照常规方式编写节点，Zenoh 的集成是透明的。以下是一个使用 Zenoh 的发布者和订阅者示例：

4.1 发布者节点

#include "rclcpp/rclcpp.hpp"
#include "std_msgs/msg/string.hpp"int main(int argc, char *argv[]) {rclcpp::init(argc, argv);auto node = rclcpp::Node::make_shared("zenoh_publisher");auto publisher = node->create_publisher<std_msgs::msg::String>("zenoh_topic", 10);rclcpp::WallRate loop_rate(1);while (rclcpp::ok()) {auto message = std_msgs::msg::String();message.data = "Hello from Zenoh Publisher!";RCLCPP_INFO(node->get_logger(), "Publishing: '%s'", message.data.c_str());publisher->publish(message);rclcpp::spin_some(node);loop_rate.sleep();}rclcpp::shutdown();return 0;
}

4.2 订阅者节点

#include "rclcpp/rclcpp.hpp"
#include "std_msgs/msg/string.hpp"void topic_callback(const std_msgs::msg::String::SharedPtr msg) {RCLCPP_INFO(rclcpp::get_logger("zenoh_subscriber"), "I heard: '%s'", msg->data.c_str());
}int main(int argc, char *argv[]) {rclcpp::init(argc, argv);auto node = rclcpp::Node::make_shared("zenoh_subscriber");auto subscription = node->create_subscription<std_msgs::msg::String>("zenoh_topic", 10, topic_callback);rclcpp::spin(node);rclcpp::shutdown();return 0;
}

这些节点编写完成后，使用常规的 ROS2 编译工具如 colcon 进行编译：

colcon build

确保在运行时，环境变量 RMW_IMPLEMENTATION 设置为 rmw_zenoh_cpp，然后可以启动发布者和订阅者：

ros2 run <package_name> zenoh_publisher
ros2 run <package_name> zenoh_subscriber

5. ROS2跨进程通信性能测试

为了评估 Zenoh 的性能，我们按照以上方法进行了ROS2跨进程通信的吞吐量和延迟测试，测试数据如下：

5.1 吞吐量测试 (throughput_in_message_per_second)

count	1kb （bytes）	2kb（bytes）	4kb（bytes）	8kb（bytes）	16kb（bytes）
1	258643.170	172798.709	119124.435	90914.272	79555.248
2	261220.364	166884.978	112133.029	87834.264	79053.264
3	252042.470	157376.734	111920.862	96717.972	78143.373
4	276926.169	156251.875	110722.081	89124.870	75229.330
5	261305.844	164742.864	118617.884	98019.081	76151.792

• 随着消息大小增加，吞吐量逐渐降低。小消息（如1KB）的传输速率最高，约为25万条/秒，而大消息（如16KB）的吞吐量则降至约7.5万条/秒。
• 消息大小对吞吐量影响显著，较大消息由于传输开销增加，导致吞吐能力下降。

5.2 延迟测试 (us)

count	1kb （us）	2kb（us）	4kb（us）	8kb（us）	16kb（us）
1	67	58	65	64	97
2	96	69	58	71	76
3	82	62	65	63	86
4	68	66	58	71	80
5	104	60	64	62	89

• 不同消息大小的延迟相对稳定。小消息（如1KB）的延迟在 67-104 微秒之间，而大消息（如16KB）的延迟在 76-97 微秒之间。
• 延迟随消息大小略有波动，但整体表现较好，能够满足大多数低延迟应用场景的需求。

结论：Zenoh 在跨进程通信中表现出较高的吞吐量和低延迟特性。对于小消息，Zenoh 可以在确保低延迟的同时实现高吞吐量。虽然大消息的吞吐量有所降低，但仍保持了较低的延迟，展示出良好的通信性能。这使得 Zenoh 适合在低延迟和高性能要求的应用中使用，尤其是在分布式系统中。

6. 参考文章

eclipse zenoh 助力雾计算和边缘计算