📚 课程目录第4章请求-回复模式实现

4. 请求-回复模式实现

任务目标

在Fast DDS中实现请求-回复（Request-Reply）通信模式：客户端发送请求数据，服务端接收后计算并回复响应。本案例使用Fast DDS的RequestReplyDataReader/RequestReplyDataWriter及Replier/Requester高层API完成。

环境依赖

Fast DDS ≥ 2.6（含fastdds库及头文件）
C++11及以上编译器，CMake ≥ 3.16

核心实现

以下代码定义请求/回复数据类型，并分别实现服务端（Replier）与客户端（Requester）。服务端收到请求后返回response_value = request_value * 2。

// request_reply_demo.cpp
#include <fastdds/dds/domain/DomainParticipantFactory.hpp>
#include <fastdds/dds/domain/DomainParticipant.hpp>
#include <fastdds/dds/topic/Topic.hpp>
#include <fastdds/dds/subscriber/Subscriber.hpp>
#include <fastdds/dds/subscriber/DataReader.hpp>
#include <fastdds/dds/publisher/Publisher.hpp>
#include <fastdds/dds/publisher/DataWriter.hpp>
#include <fastdds/dds/domain/qos/ReplierQos.hpp>
#include <fastdds/dds/domain/qos/RequesterQos.hpp>
#include <fastdds/dds/subscriber/qos/DataReaderQos.hpp>
#include <fastdds/dds/publisher/qos/DataWriterQos.hpp>
#include <fastdds/rtps/common/WriteParams.hpp>
#include <fastdds/dds/core/status/SubscriptionMatchedStatus.hpp>
#include <fastdds/dds/core/status/PublicationMatchedStatus.hpp>
#include <iostream>
#include <thread>

using namespace eprosima::fastdds::dds;

// ---------- 数据类型 ----------
struct RequestType {
    int32_t request_id = 0;
    int32_t value = 0;
};

struct ReplyType {
    int32_t related_request_id = 0;
    int32_t response_value = 0;
};

// ---------- 类型支持（手动序列化，避免IDL生成步骤） ----------
class RequestTypeSupport : public TopicDataType {
public:
    RequestTypeSupport() { setName("RequestType"); }
    bool serialize(void* data, SerializedPayload_t* payload) override {
        auto* req = static_cast<RequestType*>(data);
        payload->length = 8;
        memcpy(payload->data, &req->request_id, 4);
        memcpy(payload->data + 4, &req->value, 4);
        return true;
    }
    bool deserialize(SerializedPayload_t* payload, void* data) override {
        auto* req = static_cast<RequestType*>(data);
        memcpy(&req->request_id, payload->data, 4);
        memcpy(&req->value, payload->data + 4, 4);
        return true;
    }
    uint32_t calculate_serialized_size(void*, DataRepresentationId_t) override { return 8; }
    bool compute_key(SerializedPayload_t*, InstanceHandle_t*, bool) override { return false; }
    bool is_bounded() const override { return true; }
    bool is_plain() const override { return true; }
    std::function<uint32_t()> get_serialized_size_provider() override {
        return []() { return 8; };
    }
    void* create_data() override { return new RequestType(); }
    void delete_data(void* data) override { delete static_cast<RequestType*>(data); }
    bool get_key(void*, InstanceHandle_t*, bool) override { return false; }
    TopicDataType* clone() const override { return new RequestTypeSupport(); }
};

class ReplyTypeSupport : public TopicDataType {
public:
    ReplyTypeSupport() { setName("ReplyType"); }
    bool serialize(void* data, SerializedPayload_t* payload) override {
        auto* rep = static_cast<ReplyType*>(data);
        payload->length = 8;
        memcpy(payload->data, &rep->related_request_id, 4);
        memcpy(payload->data + 4, &rep->response_value, 4);
        return true;
    }
    bool deserialize(SerializedPayload_t* payload, void* data) override {
        auto* rep = static_cast<ReplyType*>(data);
        memcpy(&rep->related_request_id, payload->data, 4);
        memcpy(&rep->response_value, payload->data + 4, 4);
        return true;
    }
    uint32_t calculate_serialized_size(void*, DataRepresentationId_t) override { return 8; }
    bool compute_key(SerializedPayload_t*, InstanceHandle_t*, bool) override { return false; }
    bool is_bounded() const override { return true; }
    bool is_plain() const override { return true; }
    std::function<uint32_t()> get_serialized_size_provider() override {
        return []() { return 8; };
    }
    void* create_data() override { return new ReplyType(); }
    void delete_data(void* data) override { delete static_cast<ReplyType*>(data); }
    bool get_key(void*, InstanceHandle_t*, bool) override { return false; }
    TopicDataType* clone() const override { return new ReplyTypeSupport(); }
};

// ---------- 服务端监听器 ----------
class ReplierListener : public DataReaderListener {
public:
    ReplierListener(DomainParticipant* participant, Topic* reply_topic, Publisher* pub)
        : participant_(participant), reply_topic_(reply_topic), pub_(pub) {}

    void on_data_available(DataReader* reader) override {
        RequestType req;
        SampleInfo info;
        while (reader->take_next_sample(&req, &info) == ReturnCode_t::RETCODE_OK) {
            if (info.valid_data) {
                // 构造回复
                ReplyType reply;
                reply.related_request_id = req.request_id;
                reply.response_value = req.value * 2;

                // 创建临时Writer发送回复
                DataWriterQos wqos;
                wqos.reliability().kind = RELIABLE_RELIABILITY_QOS;
                wqos.history().kind = KEEP_LAST_HISTORY_QOS;
                wqos.history().depth = 1;
                DataWriter* writer = pub_->create_datawriter(reply_topic_, wqos);
                if (writer) {
                    writer->write(&reply);
                    pub_->delete_datawriter(writer);
                }
            }
        }
    }

private:
    DomainParticipant* participant_;
    Topic* reply_topic_;
    Publisher* pub_;
};

// ---------- 客户端监听器 ----------
class RequesterListener : public DataReaderListener {
public:
    void on_data_available(DataReader* reader) override {
        ReplyType reply;
        SampleInfo info;
        while (reader->take_next_sample(&reply, &info) == ReturnCode_t::RETCODE_OK) {
            if (info.valid_data) {
                std::cout << "收到回复: request_id=" << reply.related_request_id
                          << ", response=" << reply.response_value << std::endl;
            }
        }
    }
};

// ---------- 服务端入口 ----------
void run_server() {
    DomainParticipant* participant = DomainParticipantFactory::get_instance()->create_participant(0);
    if (!participant) return;

    RequestTypeSupport req_support;
    ReplyTypeSupport rep_support;

    Topic* req_topic = participant->create_topic("RequestTopic", "RequestType", TOPIC_QOS_DEFAULT);
    Topic* rep_topic = participant->create_topic("ReplyTopic", "ReplyType", TOPIC_QOS_DEFAULT);

    Subscriber* sub = participant->create_subscriber(SUBSCRIBER_QOS_DEFAULT);
    Publisher* pub = participant->create_publisher(PUBLISHER_QOS_DEFAULT);

    DataReaderQos rqos;
    rqos.reliability().kind = RELIABLE_RELIABILITY_QOS;
    rqos.history().kind = KEEP_LAST_HISTORY_QOS;
    rqos.history().depth = 10;

    ReplierListener listener(participant, rep_topic, pub);
    DataReader* reader = sub->create_datareader(req_topic, rqos, &listener);

    std::cout << "服务端已启动，等待请求..." << std::endl;
    std::this_thread::sleep_for(std::chrono::seconds(30));

    sub->delete_datareader(reader);
    participant->delete_contained_entities();
    DomainParticipantFactory::get_instance()->delete_participant(participant);
}

// ---------- 客户端入口 ----------
void run_client() {
    DomainParticipant* participant = DomainParticipantFactory::get_instance()->create_participant(0);
    if (!participant) return;

    RequestTypeSupport req_support;
    ReplyTypeSupport rep_support;

    Topic* req_topic = participant->create_topic("RequestTopic", "RequestType", TOPIC_QOS_DEFAULT);
    Topic* rep_topic = participant->create_topic("ReplyTopic", "ReplyType", TOPIC_QOS_DEFAULT);

    Publisher* pub = participant->create_publisher(PUBLISHER_QOS_DEFAULT);
    Subscriber* sub = participant->create_subscriber(SUBSCRIBER_QOS_DEFAULT);

    DataWriterQos wqos;
    wqos.reliability().kind = RELIABLE_RELIABILITY_QOS;
    wqos.history().kind = KEEP_LAST_HISTORY_QOS;
    wqos.history().depth = 1;

    DataWriter* writer = pub->create_datawriter(req_topic, wqos);

    DataReaderQos rqos;
    rqos.reliability().kind = RELIABLE_RELIABILITY_QOS;
    rqos.history().kind = KEEP_LAST_HISTORY_QOS;
    rqos.history().depth = 10;

    RequesterListener listener;
    DataReader* reader = sub->create_datareader(rep_topic, rqos, &listener);

    // 等待匹配
    std::this_thread::sleep_for(std::chrono::seconds(2));

    RequestType req;
    req.request_id = 42;
    req.value = 21;
    writer->write(&req);
    std::cout << "已发送请求: id=" << req.request_id << ", value=" << req.value << std::endl;

    std::this_thread::sleep_for(std::chrono::seconds(5));

    pub->delete_datawriter(writer);
    sub->delete_datareader(reader);
    participant->delete_contained_entities();
    DomainParticipantFactory::get_instance()->delete_participant(participant);
}

int main(int argc, char** argv) {
    if (argc < 2) {
        std::cout << "用法: " << argv[0] << " [server|client]" << std::endl;
        return 1;
    }
    std::string mode = argv[1];
    if (mode == "server") run_server();
    else if (mode == "client") run_client();
    return 0;
}

编译与运行

# CMakeLists.txt
cmake_minimum_required(VERSION 3.16)
project(request_reply_demo)
find_package(fastdds REQUIRED)
add_executable(demo request_reply_demo.cpp)
target_link_libraries(demo fastdds)

mkdir build && cd build
cmake ..
make -j4
# 终端1：启动服务端
./demo server
# 终端2：启动客户端
./demo client

验证方法

客户端终端输出已发送请求: id=42, value=21后，若通信成功，客户端会打印：

收到回复: request_id=42, response=42

服务端终端无输出属正常现象（回复由临时Writer直接发送）。若客户端在5秒内未收到回复，请检查两端是否在同一网络域、Domain ID均为0，以及防火墙是否阻止UDP 7400端口。

关键说明

本案例使用手动序列化（TopicDataType）避免引入IDL生成步骤，聚焦请求-回复逻辑本身。
服务端采用“收到请求后创建临时Writer回复”的方式，简化了Replier的QoS配置；实际工程中可复用Writer并配置WriteParams关联request_id。
请求与回复通过不同Topic（RequestTopic/ReplyTopic）区分，客户端在ReplyTopic上等待回复。