C++超低延迟通信：io_uring原生套接字vs. ZeroMQ等消息队列

• 原生io_uring套接字：性能天花板，延迟最低。直接基于内核io_uring接口操作套接字，无中间库的协议封装、线程调度或冗余内存拷贝开销，UDP场景下尾延迟可压至亚微秒到几微秒级，吞吐量也是所有选项中最高的，完全适配极致低延迟需求。
• ZeroMQ：性能尚可但存在明显开销。它封装了REQ/REP等标准化通信模式，协议层和线程模型会带来额外延迟，尾延迟约为原生io_uring的2-5倍，吞吐量低10%-30%。适合快速搭建通用通信架构，但并非极致低延迟的最优选择。
• NanoMQ：轻量但受限于MQTT协议。作为轻量化MQTT broker，延迟表现优于ZeroMQ，但MQTT协议头、主题匹配等逻辑仍会产生开销，尾延迟约为原生的1.5-3倍。适配MQTT生态下的低延迟场景，但性能上限低于原生io_uring。
• Aeron UDP：最接近原生的高层库。专为低延迟设计，采用预分配内存、直接IO、无锁队列等优化手段，延迟几乎追平原生io_uring，尾延迟仅为原生的1.1-2倍，批量传输场景下吞吐量甚至可能更高。核心优势是提供了可靠UDP实现，无需自行处理重传、排序、流量控制等复杂逻辑，性价比极高。

是否值得采用原生io_uring套接字，核心取决于你的需求和团队能力：

• 值得的场景：
  • 核心需求为极致超低延迟（亚微秒到5微秒级），且业务逻辑简单，无需复杂通信模式；
  • 团队具备资深C++网络编程经验，能处理套接字底层细节（如UDP丢包重传、TCP粘包、io_uring队列管理、内存对齐、内核版本兼容性）。
    这种场景下，原生io_uring能帮你拿到最极致的性能，规避任何不必要的开销。
• 不值得的场景：
  • 延迟要求在几十微秒级，对极致性能无硬性要求；
  • 团队精力有限，不愿投入大量时间处理底层边缘情况（如io_uring提交/完成队列死锁、内存泄漏、不同内核版本的API差异）；
  • 需要可靠传输、发布订阅/请求响应等高层通信特性。
    这种情况选择Aeron UDP或ZeroMQ更划算，高层库已封装所有复杂逻辑，开发速度快、维护成本低。

以下是基于liburing的极简UDP服务器和客户端示例，无HTTP协议依赖：

UDP服务器代码 #

#include <liburing.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <unistd.h>
#include <cstring>
#include <iostream>

#define QUEUE_DEPTH 16
#define BUF_SIZE 1024

struct conn_data {
    int sockfd;
    struct sockaddr_in client_addr;
    socklen_t addr_len;
    char buf[BUF_SIZE];
};

int main() {
    // 创建UDP套接字
    int sockfd = socket(AF_INET, SOCK_DGRAM, 0);
    if (sockfd < 0) {
        perror("socket failed");
        return 1;
    }

    // 绑定端口
    struct sockaddr_in server_addr{};
    server_addr.sin_family = AF_INET;
    server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
    server_addr.sin_port = htons(8080);
    if (bind(sockfd, (struct sockaddr*)&server_addr, sizeof(server_addr)) < 0) {
        perror("bind failed");
        close(sockfd);
        return 1;
    }

    // 初始化io_uring
    struct io_uring ring;
    if (io_uring_queue_init(QUEUE_DEPTH, &ring, 0) < 0) {
        perror("io_uring_queue_init failed");
        close(sockfd);
        return 1;
    }

    // 准备接收请求
    struct conn_data* data = new conn_data{};
    data->sockfd = sockfd;
    data->addr_len = sizeof(data->client_addr);

    struct io_uring_sqe* sqe = io_uring_get_sqe(&ring);
    io_uring_prep_recvfrom(sqe, sockfd, data->buf, BUF_SIZE, 0,
                           (struct sockaddr*)&data->client_addr, &data->addr_len);
    io_uring_sqe_set_data(sqe, data);
    io_uring_submit(&ring);

    // 循环处理完成事件
    while (true) {
        struct io_uring_cqe* cqe;
        int ret = io_uring_wait_cqe(&ring, &cqe);
        if (ret < 0) {
            perror("io_uring_wait_cqe failed");
            break;
        }

        conn_data* req_data = (conn_data*)io_uring_cqe_get_data(cqe);
        if (cqe->res < 0) {
            fprintf(stderr, "recvfrom failed: %d\n", cqe->res);
            delete req_data;
            io_uring_cqe_seen(&ring, cqe);
            continue;
        }

        // 打印收到的数据
        printf("Received %d bytes: %.*s\n", cqe->res, cqe->res, req_data->buf);

        // 回复客户端
        sqe = io_uring_get_sqe(&ring);
        io_uring_prep_sendto(sqe, req_data->sockfd, req_data->buf, cqe->res, 0,
                             (struct sockaddr*)&req_data->client_addr, req_data->addr_len);
        io_uring_sqe_set_data(sqe, req_data);
        io_uring_submit(&ring);

        io_uring_cqe_seen(&ring, cqe);
    }

    // 清理资源
    io_uring_queue_exit(&ring);
    close(sockfd);
    return 0;
}

UDP客户端代码 #

#include <liburing.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <unistd.h>
#include <cstring>
#include <iostream>

#define QUEUE_DEPTH 8
#define BUF_SIZE 1024

struct req_data {
    int sockfd;
    struct sockaddr_in server_addr;
    char send_buf[BUF_SIZE];
    char recv_buf[BUF_SIZE];
    ssize_t send_len;
};

int main() {
    // 创建UDP套接字
    int sockfd = socket(AF_INET, SOCK_DGRAM, 0);
    if (sockfd < 0) {
        perror("socket failed");
        return 1;
    }

    // 配置服务器地址
    struct sockaddr_in server_addr{};
    server_addr.sin_family = AF_INET;
    server_addr.sin_port = htons(8080);
    if (inet_pton(AF_INET, "127.0.0.1", &server_addr.sin_addr) <= 0) {
        perror("inet_pton failed");
        close(sockfd);
        return 1;
    }

    // 初始化io_uring
    struct io_uring ring;
    if (io_uring_queue_init(QUEUE_DEPTH, &ring, 0) < 0) {
        perror("io_uring_queue_init failed");
        close(sockfd);
        return 1;
    }

    // 准备发送数据
    struct req_data* data = new req_data{};
    data->sockfd = sockfd;
    data->server_addr = server_addr;
    strncpy(data->send_buf, "Hello from io_uring UDP client", BUF_SIZE-1);
    data->send_len = strlen(data->send_buf);

    struct io_uring_sqe* sqe = io_uring_get_sqe(&ring);
    io_uring_prep_sendto(sqe, sockfd, data->send_buf, data->send_len, 0,
                         (struct sockaddr*)&data->server_addr, sizeof(server_addr));
    io_uring_sqe_set_data(sqe, data);
    io_uring_submit(&ring);

    // 等待发送完成
    struct io_uring_cqe* cqe;
    int ret = io_uring_wait_cqe(&ring, &cqe);
    if (ret < 0) {
        perror("io_uring_wait_cqe failed");
        delete data;
        io_uring_queue_exit(&ring);
        close(sockfd);
        return 1;
    }

    if (cqe->res < 0) {
        fprintf(stderr, "sendto failed: %d\n", cqe->res);
        delete data;
        io_uring_cqe_seen(&ring, cqe);
        io_uring_queue_exit(&ring);
        close(sockfd);
        return 1;
    }
    printf("Sent %zd bytes\n", data->send_len);
    io_uring_cqe_seen(&ring, cqe);

    // 准备接收回复
    sqe = io_uring_get_sqe(&ring);
    io_uring_prep_recvfrom(sqe, sockfd, data->recv_buf, BUF_SIZE, 0,
                           nullptr, nullptr);
    io_uring_sqe_set_data(sqe, data);
    io_uring_submit(&ring);

    ret = io_uring_wait_cqe(&ring, &cqe);
    if (ret < 0) {
        perror("io_uring_wait_cqe failed");
        delete data;
        io_uring_queue_exit(&ring);
        close(sockfd);
        return 1;
    }

    if (cqe->res < 0) {
        fprintf(stderr, "recvfrom failed: %d\n", cqe->res);
    } else {
        printf("Received reply: %.*s\n", cqe->res, data->recv_buf);
    }

    // 清理资源
    delete data;
    io_uring_cqe_seen(&ring, cqe);
    io_uring_queue_exit(&ring);
    close(sockfd);
    return 0;
}

编译说明 #

使用gcc编译时需链接liburing库：

g++ udp_server.cpp -o udp_server -luring
g++ udp_client.cpp -o udp_client -luring

内容的提问来源于stack exchange，提问作者works