4. 使用QoS策略控制消息可靠性

任务目标

本案例将创建两个OpenDDS数据读取端（DataReader）：一个配置RELIABLE可靠性QoS，另一个配置BEST_EFFORT可靠性QoS，并让它们订阅同一个Topic。通过发布端发送一批消息，观察两种策略下消息接收的完整性差异，从而理解可靠性QoS对消息传递行为的影响。

环境与依赖

需要已安装OpenDDS及其依赖（ACE/TAO），并配置好DDS_ROOT和ACE_ROOT环境变量。本案例使用C++实现，编译需要OpenDDS的IDL编译器（opendds_idl）和对应库文件。

IDL定义

// Message.idl
module Messenger {
    struct Message {
        long index;
        string content;
    };
    #pragma keylist Message index
};

关键完整代码

// Publisher.cpp - 发布端
#include "MessageTypeSupportImpl.h"
#include <dds/DCPS/Service_Participant.h>
#include <dds/DCPS/Marked_Default_Qos.h>
#include <dds/DCPS/PublisherImpl.h>
#include <dds/DCPS/transport/framework/TransportRegistry.h>
#include <dds/DCPS/transport/simpleTCP/SimpleTcp.h>
#include <iostream>

int main(int argc, char* argv[]) {
    DDS::DomainParticipantFactory_var dpf = TheParticipantFactoryWithArgs(argc, argv);
    DDS::DomainParticipant_var participant = dpf->create_participant(
        42, PARTICIPANT_QOS_DEFAULT, 0, OpenDDS::DCPS::DEFAULT_STATUS_MASK);

    Messenger::MessageTypeSupport_var ts = new Messenger::MessageTypeSupportImpl();
    ts->register_type(participant, "");

    DDS::Topic_var topic = participant->create_topic(
        "MessageTopic", ts->get_type_name(), TOPIC_QOS_DEFAULT, 0, OpenDDS::DCPS::DEFAULT_STATUS_MASK);

    // 配置TCP传输
    OpenDDS::DCPS::TransportImpl_rch tcp = TheTransportRegistry->create_transport_impl(
        "tcp1", "SimpleTcp", OpenDDS::DCPS::DEFAULT_CONFIG_ENTRY);
    OpenDDS::DCPS::TransportConfig_rch cfg = TheTransportRegistry->create_config("pub_cfg");
    cfg->instances_.push_back(tcp);
    TheTransportRegistry->bind_config(cfg, participant);

    DDS::Publisher_var publisher = participant->create_publisher(PUBLISHER_QOS_DEFAULT, 0, OpenDDS::DCPS::DEFAULT_STATUS_MASK);
    DDS::DataWriter_var dw = publisher->create_datawriter(topic, DATAWRITER_QOS_DEFAULT, 0, OpenDDS::DCPS::DEFAULT_STATUS_MASK);
    Messenger::MessageDataWriter_var writer = Messenger::MessageDataWriter::_narrow(dw);

    // 发送20条消息
    for (int i = 0; i < 20; ++i) {
        Messenger::Message msg;
        msg.index = i;
        msg.content = "msg_" + std::to_string(i);
        writer->write(msg, DDS::HANDLE_NIL);
        std::cout << "Published: " << msg.index << std::endl;
        ACE_OS::sleep(ACE_Time_Value(0, 100000)); // 100ms间隔
    }

    ACE_OS::sleep(ACE_Time_Value(2)); // 等待消息传递
    participant->delete_contained_entities();
    dpf->delete_participant(participant);
    TheServiceParticipant->shutdown();
    return 0;
}

// Subscriber.cpp - 订阅端（同时创建RELIABLE和BEST_EFFORT两个Reader）
#include "MessageTypeSupportImpl.h"
#include <dds/DCPS/Service_Participant.h>
#include <dds/DCPS/Marked_Default_Qos.h>
#include <dds/DCPS/SubscriberImpl.h>
#include <dds/DCPS/transport/framework/TransportRegistry.h>
#include <dds/DCPS/transport/simpleTCP/SimpleTcp.h>
#include <iostream>

class ReaderListener : public DDS::DataReaderListener {
public:
    ReaderListener(const std::string& name) : name_(name), count_(0) {}
    void on_data_available(DDS::DataReader_ptr reader) {
        Messenger::MessageDataReader_var dr = Messenger::MessageDataReader::_narrow(reader);
        Messenger::Message msg;
        DDS::SampleInfo si;
        while (dr->take_next_sample(msg, si) == DDS::RETCODE_OK) {
            if (si.valid_data) {
                ++count_;
                std::cout << "[" << name_ << "] Received: " << msg.index << std::endl;
            }
        }
    }
    int count() const { return count_; }
private:
    std::string name_;
    int count_;
};

int main(int argc, char* argv[]) {
    DDS::DomainParticipantFactory_var dpf = TheParticipantFactoryWithArgs(argc, argv);
    DDS::DomainParticipant_var participant = dpf->create_participant(
        42, PARTICIPANT_QOS_DEFAULT, 0, OpenDDS::DCPS::DEFAULT_STATUS_MASK);

    Messenger::MessageTypeSupport_var ts = new Messenger::MessageTypeSupportImpl();
    ts->register_type(participant, "");

    DDS::Topic_var topic = participant->create_topic(
        "MessageTopic", ts->get_type_name(), TOPIC_QOS_DEFAULT, 0, OpenDDS::DCPS::DEFAULT_STATUS_MASK);

    OpenDDS::DCPS::TransportImpl_rch tcp = TheTransportRegistry->create_transport_impl(
        "tcp2", "SimpleTcp", OpenDDS::DCPS::DEFAULT_CONFIG_ENTRY);
    OpenDDS::DCPS::TransportConfig_rch cfg = TheTransportRegistry->create_config("sub_cfg");
    cfg->instances_.push_back(tcp);
    TheTransportRegistry->bind_config(cfg, participant);

    DDS::Subscriber_var subscriber = participant->create_subscriber(SUBSCRIBER_QOS_DEFAULT, 0, OpenDDS::DCPS::DEFAULT_STATUS_MASK);

    // 创建RELIABLE Reader
    DDS::DataReaderQos reliable_qos;
    subscriber->get_default_datareader_qos(reliable_qos);
    reliable_qos.reliability.kind = DDS::RELIABLE_RELIABILITY_QOS;
    reliable_qos.reliability.max_blocking_time.sec = 1;
    reliable_qos.reliability.max_blocking_time.nanosec = 0;

    ReaderListener* rel_listener = new ReaderListener("RELIABLE");
    DDS::DataReader_var rel_reader = subscriber->create_datareader(
        topic, reliable_qos, rel_listener, OpenDDS::DCPS::DEFAULT_STATUS_MASK);

    // 创建BEST_EFFORT Reader
    DDS::DataReaderQos be_qos;
    subscriber->get_default_datareader_qos(be_qos);
    be_qos.reliability.kind = DDS::BEST_EFFORT_RELIABILITY_QOS;

    ReaderListener* be_listener = new ReaderListener("BEST_EFFORT");
    DDS::DataReader_var be_reader = subscriber->create_datareader(
        topic, be_qos, be_listener, OpenDDS::DCPS::DEFAULT_STATUS_MASK);

    std::cout << "Subscribers created. Waiting for messages..." << std::endl;
    ACE_OS::sleep(ACE_Time_Value(10)); // 等待接收

    std::cout << "=== Results ===" << std::endl;
    std::cout << "RELIABLE received: " << rel_listener->count() << " messages" << std::endl;
    std::cout << "BEST_EFFORT received: " << be_listener->count() << " messages" << std::endl;

    participant->delete_contained_entities();
    dpf->delete_participant(participant);
    TheServiceParticipant->shutdown();
    return 0;
}

编译与运行

# 1. 生成IDL绑定代码
opendds_idl Message.idl
# 2. 编译（以Linux/g++为例）
g++ -std=c++11 -I$DDS_ROOT -I$ACE_ROOT -I. -c MessageC.cpp MessageTypeSupport.cpp MessageTypeSupportImpl.cpp
g++ -std=c++11 -I$DDS_ROOT -I$ACE_ROOT -I. Publisher.cpp MessageC.o MessageTypeSupport.o MessageTypeSupportImpl.o \
    -L$DDS_ROOT/lib -lOpenDDS_Dcps -lOpenDDS_SimpleTcp -lACE -o publisher
g++ -std=c++11 -I$DDS_ROOT -I$ACE_ROOT -I. Subscriber.cpp MessageC.o MessageTypeSupport.o MessageTypeSupportImpl.o \
    -L$DDS_ROOT/lib -lOpenDDS_Dcps -lOpenDDS_SimpleTcp -lACE -o subscriber
# 3. 运行（先启动订阅端，再启动发布端）
./subscriber &
./publisher

验证方法与判定标准

运行后观察两个终端输出：

• RELIABLE端应收到全部20条消息（编号0-19），因为可靠传输会重发丢失的消息。
• BEST_EFFORT端可能收到少于20条消息（具体数量取决于网络状况和时序），因为尽力传输不保证送达。

通过条件：RELIABLE端计数为20，BEST_EFFORT端计数≤20。若两者都收到20条，可增大消息数量或缩短发送间隔（如改为10ms）以制造更明显的差异。