当前位置：首页 > news >正文

Windows 10/11 下保姆级教程：用 Python 3.10 和 Fast DDS 2.10.0 跑通你的第一个 DDS 通信

news 2026/6/2 19:24:22

Windows 下 Python 与 Fast DDS 的零失败通信实践指南

在分布式系统开发领域，数据分发服务(DDS)正成为实时通信的首选方案。对于习惯使用Python进行快速原型开发的工程师来说，如何在Windows平台上高效利用Fast DDS的强大功能，同时避开环境配置中的各种"坑"，是掌握这项技术的关键第一步。本文将带你从零开始，用最直接的方式搭建Python与Fast DDS的通信桥梁。

1. 环境准备：构建坚实地基

1.1 安装Fast DDS核心组件

前往eProsima官网下载Fast DDS 2.10.0的Windows二进制安装包。选择与系统匹配的版本（x86或x64）后，以管理员身份运行安装程序。安装过程中有几个关键点需要注意：

勾选"Add to PATH"选项以便全局访问fastddsgen工具
记录安装目录路径（默认通常在C:\Program Files\eProsima）
确保安装完成后在命令提示符中能执行fastddsgen --version

提示：如果遇到系统权限问题，可以尝试将安装目录权限设置为当前用户完全控制。

1.2 配置SWIG接口生成器

Fast DDS的Python绑定依赖SWIG进行接口转换。从SWIG官网下载4.1.1版本的Windows二进制包，解压后将包含swig.exe的目录添加到系统PATH环境变量。验证安装：

swig -version

应输出类似信息：

SWIG Version 4.1.1 ...

2. 编译Python绑定：搭建通信桥梁

2.1 获取fastdds_python源码

从GitHub克隆Fast-DDS-python仓库：

git clone https://github.com/eProsima/Fast-DDS-python.git cd Fast-DDS-python

2.2 使用CMake配置工程

在项目根目录创建build文件夹，然后启动CMake GUI进行配置：

设置源码路径为Fast-DDS-python目录
设置构建路径为新建的build目录
点击"Configure"，选择Visual Studio 2019作为生成器
确保以下关键变量正确设置：
- Python_EXECUTABLE：指向你的Python 3.10解释器
- SWIG_EXECUTABLE：指向swig.exe路径
点击"Generate"生成VS解决方案

2.3 Visual Studio编译

打开生成的FastDDSPython.sln，在解决方案配置中选择"Release"，然后构建ALL_BUILD目标。成功编译后，你将在build目录中获得两个关键文件：

fastdds.py：Python模块接口
_fastdds_python.pyd：核心二进制扩展

将这两个文件复制到你的Python环境site-packages目录，或添加到PYTHONPATH中。

3. 创建第一个DDS应用：HelloWorld实战

3.1 定义数据结构

创建HelloWorld.idl文件定义通信数据结构：

struct HelloWorld { unsigned long index; string message; };

使用fastddsgen生成Python绑定：

fastddsgen.bat -python HelloWorld.idl

这将生成三个关键文件：

HelloWorld.py：Python数据结构定义
HelloWorld.h/.cxx：C++封装代码
HelloWorldWrapper.i：SWIG接口文件

3.2 编译数据类型支持库

使用CMake创建新工程编译生成的数据类型：

创建build目录并运行CMake
设置HelloWorld.idl生成的代码为源码路径
添加Fast DDS安装目录为依赖路径
生成并编译VS解决方案

编译成功后，你将获得_HelloWorldWrapper.pyd和HelloWorld.dll文件。

4. 实现发布/订阅模式

4.1 发布者实现

创建publisher.py：

import fastdds import HelloWorld def main(): participant = fastdds.DomainParticipant() publisher = participant.create_publisher(fastdds.PublisherQos()) topic = participant.create_topic("HelloWorldTopic", HelloWorld.HelloWorld) writer = publisher.create_datawriter(topic, fastdds.DataWriterQos()) sample = HelloWorld.HelloWorld() for i in range(10): sample.index = i sample.message = "Hello from Python!" writer.write(sample) print(f"Sent: {sample.index} - {sample.message}")

4.2 订阅者实现

创建subscriber.py：

import fastdds import HelloWorld class SubscriberListener(fastdds.DataReaderListener): def on_data_available(self, reader): info = fastdds.SampleInfo() data = HelloWorld.HelloWorld() reader.take_next_sample(data, info) print(f"Received: {data.index} - {data.message}") def main(): participant = fastdds.DomainParticipant() subscriber = participant.create_subscriber(fastdds.SubscriberQos()) topic = participant.create_topic("HelloWorldTopic", HelloWorld.HelloWorld) reader = subscriber.create_datareader(topic, fastdds.DataReaderQos(), SubscriberListener()) import time time.sleep(10) # 等待接收消息

4.3 运行测试

在两个独立的命令提示符中分别运行：

# 终端1 python publisher.py # 终端2 python subscriber.py

成功运行时，你将看到发布者发送消息和订阅者接收消息的实时输出。

5. 常见问题排查指南

5.1 环境变量配置

确保以下环境变量正确设置：

变量名	建议值	备注
PATH	包含Fast DDS bin目录	需要重启终端生效
PYTHONPATH	包含_fastdds_python.pyd所在目录	Python模块搜索路径
FastDDS_DIR	Fast DDS安装目录	CMake查找依赖用

5.2 编译错误处理

SWIG版本不匹配：严格使用SWIG 4.1.1版本
Python版本冲突：确认CMake配置中Python路径指向3.10版本
MSVC工具链问题：使用VS2019的x64 Native Tools命令提示符进行编译

5.3 运行时问题

模块导入错误：检查所有.pyd和.dll文件是否在Python可访问路径
通信失败：确保发布者和订阅者使用相同的域ID（默认0）
内存泄漏：正确释放DDS实体，或使用Python的上下文管理器

6. 进阶配置与优化

6.1 QoS策略调整

Fast DDS提供了丰富的QoS策略来控制通信行为。以下是一个可靠通信的配置示例：

qos = fastdds.DataWriterQos() qos.reliability().kind = fastdds.RELIABLE_RELIABILITY_QOS qos.history().kind = fastdds.KEEP_ALL_HISTORY_QOS qos.durability().kind = fastdds.TRANSIENT_LOCAL_DURABILITY_QOS writer = publisher.create_datawriter(topic, qos)

6.2 性能优化技巧

使用fastdds::DomainParticipantFactory::get_instance()->load_XML_profiles_file()加载预定义的XML配置
对于高频数据，考虑使用fastdds::BEST_EFFORT_RELIABILITY_QOS和适当的history设置
在Windows上启用共享内存传输可以显著提升性能：

<!-- XML配置示例 --> <transport_descriptors> <shared_memory transport_id="SharedMemoryTransport"/> </transport_descriptors>

6.3 多进程通信

Fast DDS天然支持多进程通信。只需确保：

所有进程使用相同的域ID
网络防火墙允许Fast DDS使用的端口（默认7400-7410）
共享内存传输需要进程在同一主机运行

7. 实际应用场景扩展

7.1 与ROS 2集成

Fast DDS是ROS 2的默认中间件。Python应用可以通过以下方式与ROS 2节点通信：

使用相同的接口定义（IDL转换为ROS 2 msg/srv）
配置相同的域ID（默认0）
注意ROS 2使用的Fast DDS版本兼容性

7.2 分布式系统监控

利用Fast DDS的内置发现协议，可以构建系统监控工具：

participant = fastdds.DomainParticipant() builtin = participant.get_builtin_subscriber() datareader = builtin.get_datareader("DCPSParticipant") # 解析发现数据获取参与节点信息