3. DDS服务质量（QoS）配置实践

任务目标

在同一个 DDS 域内创建发布者与订阅者，通过修改 Reliability（可靠性）、Durability（持久性）、History（历史记录） 三类 QoS 策略，观察它们对消息投递行为的影响。核心验证点：

• Reliability 设为 BEST_EFFORT 时，晚启动的订阅者收不到之前已发送的消息；设为 RELIABLE 时，能收到重传的消息。
• Durability 设为 TRANSIENT_LOCAL 时，晚启动的订阅者能收到发布者保留的历史样本。
• History 的 depth 决定发布者/订阅者缓存保留的样本数量。

环境与依赖

• eProsima Fast DDS 2.x（含 fastdds 库与 fastrtpsgen 或 IDL 编译工具）
• C++11 及以上编译器，CMake 3.10+
• IDL 定义的数据类型 HelloWorld（含 index 与 message 字段）

QoS 策略与通信行为关系

核心实现

下面代码创建发布者与订阅者，通过命令行参数切换 QoS 组合。发布者先发 5 条消息，等待 2 秒后订阅者才启动，用于观察 Durability 与 Reliability 的差异。

#include <fastdds/dds/domain/DomainParticipantFactory.hpp>
#include <fastdds/dds/domain/DomainParticipant.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/dds/topic/Topic.hpp>
#include <fastdds/dds/topic/TypeSupport.hpp>
#include <fastdds/dds/core/policy/QosPolicies.hpp>
#include "HelloWorldPubSubTypes.hpp"
#include <iostream>
#include <thread>
#include <chrono>

using namespace eprosima::fastdds::dds;

int main(int argc, char** argv) {
    bool is_pub = (argc > 1 && std::string(argv[1]) == "pub");
    bool reliable = (argc > 2 && std::string(argv[2]) == "reliable");
    bool transient = (argc > 3 && std::string(argv[3]) == "transient");
    int depth = (argc > 4) ? std::atoi(argv[4]) : 3;

    DomainParticipant* participant =
        DomainParticipantFactory::get_instance()->create_participant(0, PARTICIPANT_QOS_DEFAULT);
    if (!participant) { std::cerr << "participant fail\n"; return 1; }

    TypeSupport type(new HelloWorldPubSubType());
    type.register_type(participant);
    Topic* topic = participant->create_topic("HelloTopic", type.get_type_name(), TOPIC_QOS_DEFAULT);

    if (is_pub) {
        Publisher* pub = participant->create_publisher(PUBLISHER_QOS_DEFAULT);
        DataWriterQos wqos = DATAWRITER_QOS_DEFAULT;
        wqos.reliability().kind = reliable ? RELIABLE_RELIABILITY_QOS : BEST_EFFORT_RELIABILITY_QOS;
        wqos.durability().kind = transient ? TRANSIENT_LOCAL_DURABILITY_QOS : VOLATILE_DURABILITY_QOS;
        wqos.history().kind = KEEP_LAST_HISTORY_QOS;
        wqos.history().depth = depth;
        DataWriter* writer = pub->create_datawriter(topic, wqos);

        HelloWorld sample;
        for (int i = 0; i < 5; ++i) {
            sample.index(i);
            sample.message("msg-" + std::to_string(i));
            writer->write(&sample);
            std::cout << "[pub] sent index=" << i << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(200));
        }
        std::cout << "[pub] waiting 5s for late subscriber...\n";
        std::this_thread::sleep_for(std::chrono::seconds(5));
        pub->delete_datawriter(writer);
        participant->delete_publisher(pub);
    } else {
        Subscriber* sub = participant->create_subscriber(SUBSCRIBER_QOS_DEFAULT);
        DataReaderQos rqos = DATAREADER_QOS_DEFAULT;
        rqos.reliability().kind = reliable ? RELIABLE_RELIABILITY_QOS : BEST_EFFORT_RELIABILITY_QOS;
        rqos.durability().kind = transient ? TRANSIENT_LOCAL_DURABILITY_QOS : VOLATILE_DURABILITY_QOS;
        rqos.history().kind = KEEP_LAST_HISTORY_QOS;
        rqos.history().depth = depth;
        DataReader* reader = sub->create_datareader(topic, rqos);

        std::cout << "[sub] started, waiting for samples...\n";
        HelloWorld data;
        SampleInfo info;
        auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(8);
        while (std::chrono::steady_clock::now() < deadline) {
            if (reader->take_next_sample(&data, &info) == ReturnCode_t::RETCODE_OK) {
                if (info.valid_data)
                    std::cout << "[sub] received index=" << data.index()
                              << " msg=" << data.message() << std::endl;
            }
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
        }
        sub->delete_datareader(reader);
        participant->delete_subscriber(sub);
    }

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

编译与运行

# 生成类型支持（若尚未生成）
fastrtpsgen -example CMake HelloWorld.idl

# 编译
mkdir build && cd build
cmake .. && make

# 终端 A：发布者，RELIABLE + TRANSIENT_LOCAL，depth=3
./qos_demo pub reliable transient 3

# 终端 B：等发布者打印 "waiting 5s" 后再启动订阅者
./qos_demo sub reliable transient 3

验证方法与判定标准

1. BEST_EFFORT + VOLATILE：发布者先发完 5 条，订阅者后启动 → 订阅者收不到任何消息（无重传、无历史）。
2. RELIABLE + VOLATILE：订阅者后启动 → 仍收不到已发消息，但若订阅者先启动，则 5 条全部到达。
3. RELIABLE + TRANSIENT_LOCAL + depth=3：订阅者后启动 → 收到最后 3 条（index=2,3,4），验证 History depth 与 Durability 共同作用。
4. RELIABLE + TRANSIENT_LOCAL + depth=5：订阅者后启动 → 收到全部 5 条。

通过条件：订阅者打印的 received index= 序列与上述预期一致，且无崩溃或超时无输出。

常见问题

• 订阅者收不到消息：检查双方 Reliability 是否兼容（RELIABLE 订阅者可匹配 BEST_EFFORT 发布者，反之不匹配）。
• TRANSIENT_LOCAL 无效：确认发布者 DataWriter 的 durability 已设为 TRANSIENT_LOCAL，且发布者未在订阅者启动前销毁。
• depth 不生效：History 需配合 KEEP_LAST 使用，KEEP_ALL 时 depth 被忽略。