📚 课程目录第4章多通道并行传输

4. 多通道并行传输

任务目标

在DDS中，单个Reliable通道的吞吐量受限于确认应答往返时间与发送窗口大小。本案例利用DDS的多通道(Multi-Channel)特性，将同一Topic的数据按Key分片到多个独立的Best-Effort通道并行发送，以提升端到端吞吐量。完成目标：实现一个发布端将数据分片到4个通道并发发送，订阅端从4个通道聚合接收并重组数据的完整可运行示例。

环境与依赖

操作系统：Ubuntu 20.04+ / Windows 10+（跨平台）
依赖库：Eclipse Cyclone DDS（需安装，提供C语言API）
编译工具：CMake 3.10+，C编译器（gcc/msvc）
安装Cyclone DDS（Ubuntu示例）：sudo apt install cyclone-dds 或从源码编译安装

实现原理

DDS标准中的MultiChannel扩展允许一个DataWriter将不同Key的数据路由到不同的底层通道。本实现使用Cyclone DDS的dds_writer_set_multicast接口（实际为dds_entity_set_qos配合DDS_PARTITION_QOS_POLICY）——更准确地说，我们通过为每个通道创建独立的Writer并绑定到不同的Partition来实现并行通道效果。每个Writer使用Best-Effort可靠性（不等待ACK），数据按Key哈希分配到对应通道，从而并行发送。

完整代码

// multi_channel_pubsub.c
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <assert.h>
#include "dds/dds.h"

#define NUM_CHANNELS 4
#define NUM_SAMPLES 1000
#define PAYLOAD_SIZE 1024

// Topic 类型定义
typedef struct {
    int32_t key;       // 用于分片路由
    int32_t seq;       // 序列号，用于重组
    uint8_t data[PAYLOAD_SIZE];
} Chunk;

// 生成并发布数据
static void publish(dds_entity_t writer, int channel_id) {
    Chunk sample;
    memset(&sample, 0, sizeof(sample));
    for (int i = channel_id; i < NUM_SAMPLES; i += NUM_CHANNELS) {
        sample.key = channel_id;
        sample.seq = i;
        // 填充数据（示例：写入通道ID和序列号）
        memcpy(sample.data, &i, sizeof(i));
        dds_return_t rc = dds_write(writer, &sample);
        if (rc != DDS_RETCODE_OK) {
            fprintf(stderr, "Write failed on channel %d: %d\n", channel_id, rc);
            exit(1);
        }
    }
}

int main(int argc, char** argv) {
    dds_entity_t participant, topic, pub, sub;
    dds_entity_t writers[NUM_CHANNELS], readers[NUM_CHANNELS];
    dds_qos_t *qos;
    dds_return_t rc;

    // 创建参与者
    participant = dds_create_participant(DDS_DOMAIN_DEFAULT, NULL, NULL);
    if (participant < 0) { fprintf(stderr, "create participant failed\n"); return 1; }

    // 注册类型
    dds_entity_t type = dds_create_type(participant, (dds_type_meta_t*)&Chunk_desc, "Chunk");
    assert(type > 0);

    // 创建Topic
    topic = dds_create_topic(participant, &Chunk_desc, "MultiChannelData", NULL, NULL);
    assert(topic > 0);

    // 创建发布者/订阅者
    pub = dds_create_publisher(participant, NULL, NULL);
    sub = dds_create_subscriber(participant, NULL, NULL);
    assert(pub > 0 && sub > 0);

    // 为每个通道创建Writer和Reader，使用不同Partition隔离
    char partition_name[32];
    for (int i = 0; i < NUM_CHANNELS; i++) {
        snprintf(partition_name, sizeof(partition_name), "chan_%d", i);
        
        // 配置QoS：Best-Effort + Partition
        qos = dds_create_qos();
        dds_qset_reliability(qos, DDS_RELIABILITY_BEST_EFFORT, 0);
        dds_qset_partition(qos, 1, &partition_name);
        
        writers[i] = dds_create_writer(pub, topic, qos, NULL);
        readers[i] = dds_create_reader(sub, topic, qos, NULL);
        dds_delete_qos(qos);
        assert(writers[i] > 0 && readers[i] > 0);
    }

    // 匹配等待
    dds_sleepfor(DDS_MSECS(2000));

    // 发布：每个通道独立线程（此处简化为顺序调用，实际可并行）
    // 注意：为演示并行，这里用4个线程模拟
    #ifdef _WIN32
        // Windows线程代码略（可用CreateThread）
    #else
        pthread_t threads[NUM_CHANNELS];
        for (int i = 0; i < NUM_CHANNELS; i++) {
            pthread_create(&threads[i], NULL, (void*(*)(void*))publish, (void*)writers[i]);
        }
        for (int i = 0; i < NUM_CHANNELS; i++) {
            pthread_join(threads[i], NULL);
        }
    #endif

    // 订阅端接收并统计
    int received[NUM_CHANNELS] = {0};
    int total_received = 0;
    dds_entity_t cond = dds_create_readcondition(readers[0], DDS_ANY_STATE);
    // 简化：轮询所有reader
    for (int round = 0; round < 100; round++) {
        for (int i = 0; i < NUM_CHANNELS; i++) {
            void *samples[NUM_SAMPLES];
            dds_sample_info_t infos[NUM_SAMPLES];
            rc = dds_take(readers[i], samples, infos, NUM_SAMPLES, DDS_ANY_STATE);
            if (rc > 0) {
                for (int j = 0; j < rc; j++) {
                    if (infos[j].valid_data) {
                        Chunk *chunk = (Chunk*)samples[j];
                        received[chunk->key]++;
                        total_received++;
                    }
                }
                dds_return_loan(readers[i], samples, rc);
            }
        }
        if (total_received >= NUM_SAMPLES) break;
        dds_sleepfor(DDS_MSECS(100));
    }

    // 验证结果
    printf("Total received: %d / %d\n", total_received, NUM_SAMPLES);
    for (int i = 0; i < NUM_CHANNELS; i++) {
        printf("Channel %d: %d samples\n", i, received[i]);
    }
    if (total_received == NUM_SAMPLES) {
        printf("PASS: All samples received correctly.\n");
    } else {
        printf("FAIL: Missing %d samples.\n", NUM_SAMPLES - total_received);
    }

    // 清理
    for (int i = 0; i < NUM_CHANNELS; i++) {
        dds_delete(writers[i]);
        dds_delete(readers[i]);
    }
    dds_delete(cond);
    dds_delete(pub);
    dds_delete(sub);
    dds_delete(topic);
    dds_delete(participant);
    return 0;
}

// 类型支持代码（由idlc生成，此处为简化手写）
const dds_topic_descriptor_t Chunk_desc = {
    .m_size = sizeof(Chunk),
    .m_align = 8,
    .m_flag = DDS_TOPIC_NO_OPTIMIZE,
    .m_nkeys = 1,
    .m_keys = (uint32_t[]){ offsetof(Chunk, key) },
    .m_nops = 3,
    .m_ops = (uint32_t[]){
        DDS_OP_ADR | DDS_OP_TYPE_4BY | DDS_OP_FLAG_SGN, offsetof(Chunk, key),
        DDS_OP_ADR | DDS_OP_TYPE_4BY | DDS_OP_FLAG_SGN, offsetof(Chunk, seq),
        DDS_OP_ADR | DDS_OP_TYPE_1BY | DDS_OP_FLAG_BASE, offsetof(Chunk, data)
    }
};

编译与运行

# 编译（需安装cyclonedds）
gcc -o multi_channel multi_channel_pubsub.c -lddsc -lpthread

# 运行（单进程内同时发布和订阅）
./multi_channel

验证方法

运行程序后观察输出：

通过条件：输出Total received: 1000 / 1000且每个通道接收数量接近250（因分片均匀），最终显示PASS。
吞吐量对比：可修改代码中NUM_CHANNELS为1（单通道）对比耗时，验证多通道并行带来的吞吐提升。
注意事项：由于使用Best-Effort，极端网络丢包可能导致少量数据丢失，本机回环测试通常无丢包。

关键说明

本实现通过Partition隔离模拟多通道，每个Partition对应一个独立通道，Writer/Reader绑定相同Partition即可通信。
实际DDS MultiChannel特性（同一Writer多通道）在Cyclone DDS中支持有限，本方法为跨实现兼容的通用方案。
发布端使用多线程并行写入不同Writer，以模拟真实并行发送；若需严格串行可去掉线程。
数据重组依据seq字段，接收端可据此排序恢复原始顺序。