4. DDS UDP单播通信

任务目标

本案例使用Fast DDS在单个主机上通过UDP单播实现DDS发布/订阅通信。发布端周期发送温度数据，订阅端接收并打印，以此验证DDS在UDP单播模式下的基本数据通路。

环境依赖

需要安装Fast DDS及其依赖（如Asio、TinyXML2等）。在Ubuntu上可通过以下命令安装：

sudo apt install libfastdds-dev

完整实现

以下代码包含一个可编译运行的完整示例。发布端和订阅端在同一可执行文件中通过命令行参数区分启动。

// dds_udp_unicast.cpp
#include <fastdds/dds/domain/DomainParticipantFactory.hpp>
#include <fastdds/dds/domain/DomainParticipant.hpp>
#include <fastdds/dds/topic/TypeSupport.hpp>
#include <fastdds/dds/publisher/Publisher.hpp>
#include <fastdds/dds/publisher/DataWriter.hpp>
#include <fastdds/dds/subscriber/Subscriber.hpp>
#include <fastdds/dds/subscriber/DataReader.hpp>
#include <fastdds/dds/subscriber/SampleInfo.hpp>
#include <fastdds/rtps/transport/UDPv4TransportDescriptor.hpp>
#include <iostream>
#include <thread>
#include <chrono>

using namespace eprosima::fastdds::dds;

// 定义温度数据类型
class Temperature
{
public:
    float value = 0.0f;
};

// 类型支持类（简化版，实际需用fastddsgen生成）
class TemperaturePubSubType : public TopicDataType
{
public:
    TemperaturePubSubType()
    {
        setName("Temperature");
        m_typeSize = sizeof(float);
    }
    bool serialize(void* data, SerializedPayload_t* payload) override
    {
        Temperature* t = static_cast<Temperature*>(data);
        memcpy(payload->data, &t->value, sizeof(float));
        payload->length = sizeof(float);
        return true;
    }
    bool deserialize(SerializedPayload_t* payload, void* data) override
    {
        Temperature* t = static_cast<Temperature*>(data);
        memcpy(&t->value, payload->data, sizeof(float));
        return true;
    }
    uint32_t calculate_serialized_size(void*, DataRepresentationId_t) override { return sizeof(float); }
    void* createData() override { return new Temperature(); }
    void deleteData(void* data) override { delete static_cast<Temperature*>(data); }
    bool getKey(void*, InstanceHandle_t*, bool) override { return false; }
};

// 订阅端监听器
class SubListener : public DataReaderListener
{
public:
    void on_data_available(DataReader* reader) override
    {
        Temperature temp;
        SampleInfo info;
        if (reader->take_next_sample(&temp, &info) == ReturnCode_t::RETCODE_OK)
        {
            if (info.valid_data)
            {
                std::cout << "收到温度: " << temp.value << " °C" << std::endl;
            }
        }
    }
};

int main(int argc, char** argv)
{
    if (argc < 2)
    {
        std::cout << "用法: " << argv[0] << " [pub|sub]" << std::endl;
        return 1;
    }
    bool is_pub = (std::string(argv[1]) == "pub");

    // 创建参与者
    DomainParticipantQos pqos = PARTICIPANT_QOS_DEFAULT;
    pqos.transport().user_transports.clear();
    auto udp_transport = std::make_shared<eprosima::fastdds::rtps::UDPv4TransportDescriptor>();
    udp_transport->sendBufferSize = 65536;
    udp_transport->receiveBufferSize = 65536;
    pqos.transport().user_transports.push_back(udp_transport);
    pqos.transport().use_builtin_transports = false;

    DomainParticipant* participant = DomainParticipantFactory::get_instance()->create_participant(0, pqos);
    if (!participant) { std::cerr << "创建参与者失败" << std::endl; return 1; }

    // 注册类型
    TemperaturePubSubType type;
    TypeSupport ts(&type);
    ts.register_type(participant);

    // 创建主题
    Topic* topic = participant->create_topic("TemperatureTopic", "Temperature", TOPIC_QOS_DEFAULT);
    if (!topic) { std::cerr << "创建主题失败" << std::endl; return 1; }

    if (is_pub)
    {
        // 发布端
        Publisher* pub = participant->create_publisher(PUBLISHER_QOS_DEFAULT);
        DataWriter* writer = pub->create_datawriter(topic, DATAWRITER_QOS_DEFAULT);
        Temperature temp;
        temp.value = 25.0f;
        for (int i = 0; i < 10; ++i)
        {
            temp.value += 0.5f;
            writer->write(&temp);
            std::cout << "发布温度: " << temp.value << " °C" << std::endl;
            std::this_thread::sleep_for(std::chrono::seconds(1));
        }
    }
    else
    {
        // 订阅端
        Subscriber* sub = participant->create_subscriber(SUBSCRIBER_QOS_DEFAULT);
        SubListener listener;
        DataReader* reader = sub->create_datareader(topic, DATAREADER_QOS_DEFAULT, &listener);
        std::cout << "订阅端启动，等待数据..." << std::endl;
        std::this_thread::sleep_for(std::chrono::seconds(15));
    }

    participant->delete_contained_entities();
    DomainParticipantFactory::get_instance()->delete_participant(participant);
    return 0;
}

编译与运行

编译命令：

g++ -std=c++11 dds_udp_unicast.cpp -o dds_demo -lfastdds -lfastcdr

在两个终端分别运行：

# 终端1（订阅端）
./dds_demo sub

# 终端2（发布端）
./dds_demo pub

验证方法

观察订阅端终端是否打印出与发布端相同序列的温度值（25.5°C、26.0°C……30.0°C）。若订阅端在15秒内收到全部10条数据，且数值与发布端一致，则说明UDP单播DDS通信成功。

关键说明

本示例显式配置了UDPv4传输并禁用内置传输，确保通信走UDP单播。实际工程中数据类型通常由fastddsgen工具从IDL文件生成，此处为聚焦通信流程而简化了类型支持类的实现。