2. 编写第一个发布-订阅程序

任务目标

本案例将使用OpenDDS实现一个最小可运行的发布-订阅程序：发布端周期发送字符串消息，订阅端接收并打印。通过该闭环理解OpenDDS的基本通信流程：域参与者 → 主题 → 发布者/订阅者 → 数据写入器/读取器 → 数据样本。

环境准备

需要OpenDDS及其依赖（ACE/TAO）已正确安装，并设置环境变量：

export DDS_ROOT=/path/to/OpenDDS
export ACE_ROOT=/path/to/ACE_wrappers
export TAO_ROOT=$ACE_ROOT/TAO
export LD_LIBRARY_PATH=$DDS_ROOT/lib:$ACE_ROOT/lib
source $DDS_ROOT/setenv.sh

IDL定义与生成

创建 Message.idl，定义消息类型：

module Messenger {
    struct Message {
        string text;
        long subject_id;
    };
    #pragma keylist Message subject_id
};

使用 opendds_idl 生成绑定代码：

opendds_idl Message.idl
tao_idl -I$DDS_ROOT -I$TAO_ROOT/tao Message.idl
tao_idl -I$DDS_ROOT -I$TAO_ROOT/tao MessageTypeSupport.idl
tao_idl -I$DDS_ROOT -I$TAO_ROOT/tao MessageDataReaderListener.idl
tao_idl -I$DDS_ROOT -I$TAO_ROOT/tao MessageDataWriterListener.idl

关键完整代码

以下为发布端与订阅端的最小实现，保存在 publisher.cpp 和 subscriber.cpp 中。两者共用同一IDL生成的头文件。

// publisher.cpp — 发布端
#include "MessageTypeSupportImpl.h"
#include <dds/DCPS/Service_Participant.h>
#include <dds/DCPS/Marked_Default_Qos.h>
#include <dds/DCPS/PublisherImpl.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, 0);
    if (!participant) { std::cerr << "create_participant failed\n"; return 1; }

    Messenger::MessageTypeSupport_var ts =
        new Messenger::MessageTypeSupportImpl();
    if (ts->register_type(participant, "") != DDS::RETCODE_OK) return 1;

    CORBA::String_var type_name = ts->get_type_name();
    DDS::Topic_var topic = participant->create_topic(
        "MessageTopic", type_name, TOPIC_QOS_DEFAULT, 0, 0);
    if (!topic) { std::cerr << "create_topic failed\n"; return 1; }

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

    Messenger::Message msg;
    msg.subject_id = 1;
    for (int i = 0; i < 5; ++i) {
        msg.text = CORBA::string_dup("Hello OpenDDS #");
        msg.text += std::to_string(i).c_str();
        writer->write(msg, DDS::HANDLE_NIL);
        std::cout << "Published: " << msg.text << std::endl;
        ACE_OS::sleep(1);
    }

    participant->delete_contained_entities();
    dpf->delete_participant(participant);
    TheServiceParticipant->shutdown();
    return 0;
}
// subscriber.cpp — 订阅端
#include "MessageTypeSupportImpl.h"
#include "MessageDataReaderListenerImpl.h"
#include <dds/DCPS/Service_Participant.h>
#include <dds/DCPS/Marked_Default_Qos.h>
#include <dds/DCPS/SubscriberImpl.h>
#include <iostream>

class ListenerImpl : public virtual OpenDDS::DCPS::LocalObject<
    DDS::DataReaderListener> {
public:
    void on_data_available(DDS::DataReader_ptr reader) override {
        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)
                std::cout << "Received: " << msg.text << std::endl;
        }
    }
    // 其他监听回调为空实现
    void on_requested_deadline_missed(DDS::DataReader_ptr, const DDS::RequestedDeadlineMissedStatus&) override {}
    void on_requested_incompatible_qos(DDS::DataReader_ptr, const DDS::RequestedIncompatibleQosStatus&) override {}
    void on_sample_rejected(DDS::DataReader_ptr, const DDS::SampleRejectedStatus&) override {}
    void on_liveliness_changed(DDS::DataReader_ptr, const DDS::LivelinessChangedStatus&) override {}
    void on_subscription_matched(DDS::DataReader_ptr, const DDS::SubscriptionMatchedStatus&) override {}
    void on_sample_lost(DDS::DataReader_ptr, const DDS::SampleLostStatus&) override {}
};

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, 0);
    if (!participant) { std::cerr << "create_participant failed\n"; return 1; }

    Messenger::MessageTypeSupport_var ts =
        new Messenger::MessageTypeSupportImpl();
    ts->register_type(participant, "");
    CORBA::String_var type_name = ts->get_type_name();
    DDS::Topic_var topic = participant->create_topic(
        "MessageTopic", type_name, TOPIC_QOS_DEFAULT, 0, 0);

    DDS::Subscriber_var sub = participant->create_subscriber(
        SUBSCRIBER_QOS_DEFAULT, 0, 0);
    ListenerImpl* listener = new ListenerImpl();
    DDS::DataReader_var dr = sub->create_datareader(
        topic, DATAREADER_QOS_DEFAULT, listener,
        DDS::DATA_AVAILABLE_STATUS);

    std::cout << "Subscriber waiting... press Ctrl+C to exit" << std::endl;
    while (true) { ACE_OS::sleep(1); }
    return 0;
}

编译与运行

创建 CMakeLists.txt 或使用以下g++命令（需根据实际安装路径调整）：

g++ -o publisher publisher.cpp MessageC.cpp MessageTypeSupportC.cpp \
    MessageDataReaderListenerC.cpp MessageDataWriterListenerC.cpp \
    -I$DDS_ROOT -I$TAO_ROOT -I$DDS_ROOT/../ACE_wrappers \
    -L$DDS_ROOT/lib -lOpenDDS_Dcps -lTAO -lACE
g++ -o subscriber subscriber.cpp MessageC.cpp MessageTypeSupportC.cpp \
    MessageDataReaderListenerC.cpp MessageDataWriterListenerC.cpp \
    -I$DDS_ROOT -I$TAO_ROOT -I$DDS_ROOT/../ACE_wrappers \
    -L$DDS_ROOT/lib -lOpenDDS_Dcps -lTAO -lACE

运行（先启动订阅端，再启动发布端）：

./subscriber &
./publisher

验证方法

判定通过条件：

  1. 订阅端控制台输出5条 Received: Hello OpenDDS #0 至 #4，与发布端发送内容一致；
  2. 发布端输出5条 Published: 日志，无错误退出；
  3. 两端进程均无崩溃或DDS错误信息。

若订阅端先启动后无输出，检查两端 domain_id（均为42）和 topic name（均为"MessageTopic"）是否一致。

通信流程示意

DomainParticipant (domain 42) Topic "MessageTopic" Publisher/Writer write(Message) Subscriber/Reader take_next_sample() DDS数据分发（字符串消息）

上图展示了本案例的核心对象关系：发布端与订阅端通过相同的域ID和主题名称建立逻辑连接，数据从Writer流向Reader，无需直接网络编程。

← 上一章 📖 返回目录 下一章 →