nanomsg / NNG 消息库完全教程 / 第 6 章：可扩展性与性能

// 启用 TCP_NODELAY
int nodelay = 1;
nng_setopt(sock, NNG_OPT_TCP_NODELAY, &nodelay, sizeof(nodelay));

// 启用 TCP Keepalive
int keepalive = 1;
nng_setopt(sock, NNG_OPT_TCP_KEEPALIVE, &keepalive, sizeof(keepalive));

# 查看当前限制
ulimit -n

# 临时增加（当前 shell）
ulimit -n 65536

# 永久增加（/etc/security/limits.conf）
# *  soft  nofile  65536
# *  hard  nofile  65536

# sysctl 调优
sysctl -w net.core.somaxconn=65535
sysctl -w net.ipv4.tcp_max_syn_backlog=65535

// NNG 默认使用 4 个 I/O 线程
// 可在程序启动前设置环境变量
// NNG_NUM_TASKQ_THREADS=8

// 或在代码中设置
// (NNG 内部使用 nng_taskq_set_threads 但非公开 API)

export NNG_NUM_TASKQ_THREADS=8
./myapp

// 减小缓冲区队列
int bufsize = 16;  // 默认 128
nng_setopt(sock, NNG_OPT_RECVBUF, &bufsize, sizeof(bufsize));
nng_setopt(sock, NNG_OPT_SENDBUF, &bufsize, sizeof(bufsize));

// 限制最大接收消息大小
size_t maxsz = 4096;  // 默认 1MB
nng_setopt(sock, NNG_OPT_RECVMAXSZ, &maxsz, sizeof(maxsz));

// nanomsg 零拷贝
void *msg = nn_allocmsg(1024, 0);
memcpy(msg, data, 1024);
nn_send(sock, &msg, NN_MSG, 0);
// 发送后不要释放 msg

// NNG 零拷贝
nng_msg *msg;
nng_msg_alloc(&msg, 1024);
memcpy(nng_msg_body(msg), data, 1024);
nng_sendmsg(sock, msg, 0);
// 发送后不要释放 msg

#define CHUNK_SIZE 65536  // 64KB

int send_large_message(nng_socket sock, const void *data, size_t len) {
    size_t offset = 0;
    while (offset < len) {
        size_t chunk = (len - offset > CHUNK_SIZE) ? CHUNK_SIZE : (len - offset);

        // 构造消息: [4字节总长][4字节偏移][数据]
        nng_msg *msg;
        nng_msg_alloc(&msg, 8 + chunk);

        void *body = nng_msg_body(msg);
        uint32_t total = (uint32_t)len;
        uint32_t off = (uint32_t)offset;
        memcpy(body, &total, 4);
        memcpy(body + 4, &off, 4);
        memcpy(body + 8, data + offset, chunk);

        int rv = nng_sendmsg(sock, msg, 0);
        if (rv != 0) return rv;

        offset += chunk;
    }
    return 0;
}

# 终端 1：启动服务端
nngcat --rep --listen tcp://*:5555 --count 10000

# 终端 2：运行客户端（发送 10000 条消息）
time nngcat --req --dial tcp://localhost:5555 \
    --data "Hello" --count 10000 --interval 0

// bench_latency.c —— REQ/REP 往返延迟测试
#include <nng/nng.h>
#include <nng/protocol/reqrep0/req.h>
#include <nng/protocol/reqrep0/rep.h>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <sys/time.h>
#include <pthread.h>

#define MSG_SIZE    64
#define ITERATIONS  10000

static double now_ms(void) {
    struct timeval tv;
    gettimeofday(&tv, NULL);
    return tv.tv_sec * 1000.0 + tv.tv_usec / 1000.0;
}

void *server_thread(void *arg) {
    nng_socket sock;
    nng_rep0_open(&sock);
    nng_listen(sock, "inproc://bench", NULL, 0);

    for (int i = 0; i < ITERATIONS; i++) {
        char buf[MSG_SIZE];
        size_t sz = sizeof(buf);
        nng_recv(sock, buf, &sz, 0);
        nng_send(sock, buf, sz, 0);
    }

    nng_close(sock);
    return NULL;
}

int main() {
    pthread_t srv;
    pthread_create(&srv, NULL, server_thread, NULL);

    // 等待服务端就绪
    usleep(100000);

    nng_socket sock;
    nng_req0_open(&sock);
    nng_dial(sock, "inproc://bench", NULL, 0);

    char data[MSG_SIZE];
    memset(data, 'A', MSG_SIZE);

    double start = now_ms();

    for (int i = 0; i < ITERATIONS; i++) {
        nng_send(sock, data, MSG_SIZE, 0);
        char buf[MSG_SIZE];
        size_t sz = sizeof(buf);
        nng_recv(sock, buf, &sz, 0);
    }

    double elapsed = now_ms() - start;

    printf("Iterations:  %d\n", ITERATIONS);
    printf("Message size: %d bytes\n", MSG_SIZE);
    printf("Total time:  %.2f ms\n", elapsed);
    printf("RTT avg:     %.3f μs\n", elapsed * 1000.0 / ITERATIONS);
    printf("Throughput:  %.0f msg/s\n", ITERATIONS / (elapsed / 1000.0));

    nng_close(sock);
    pthread_join(srv, NULL);
    return 0;
}

cc bench_latency.c -lnng -lpthread -o bench_latency
./bench_latency

// bench_throughput.c —— PUB/SUB 吞吐量测试
#include <nng/nng.h>
#include <nng/protocol/pubsub0/pub.h>
#include <nng/protocol/pubsub0/sub.h>
#include <stdio.h>
#include <string.h>
#include <sys/time.h>
#include <pthread.h>

#define MSG_COUNT  100000
#define MSG_SIZE   256

static double now_ms(void) {
    struct timeval tv;
    gettimeofday(&tv, NULL);
    return tv.tv_sec * 1000.0 + tv.tv_usec / 1000.0;
}

void *publisher(void *arg) {
    nng_socket *sock = (nng_socket *)arg;

    // 等待订阅者连接
    usleep(500000);

    char data[MSG_SIZE];
    memset(data, 'B', MSG_SIZE);

    double start = now_ms();

    for (int i = 0; i < MSG_COUNT; i++) {
        nng_send(*sock, data, MSG_SIZE, 0);
    }

    double elapsed = now_ms() - start;
    printf("Published %d messages in %.2f ms\n", MSG_COUNT, elapsed);
    printf("Throughput: %.0f msg/s (%.2f MB/s)\n",
           MSG_COUNT / (elapsed / 1000.0),
           (double)MSG_COUNT * MSG_SIZE / elapsed / 1000.0);

    return NULL;
}

int main() {
    nng_socket pub, sub;
    pthread_t pub_thread;

    nng_pub0_open(&pub);
    nng_sub0_open(&sub);

    nng_setopt_ms(sub, NNG_OPT_RECVTIMEO, 10000);
    nng_setopt(sub, NNG_OPT_SUB_SUBSCRIBE, "", 0);

    nng_listen(pub, "inproc://throughput", NULL, 0);
    nng_dial(sub, "inproc://throughput", NULL, 0);

    pthread_create(&pub_thread, NULL, publisher, &pub);

    int received = 0;
    char buf[MSG_SIZE];
    size_t sz;

    while (received < MSG_COUNT) {
        sz = sizeof(buf);
        if (nng_recv(sub, buf, &sz, 0) == 0) {
            received++;
        }
    }

    printf("Received %d messages\n", received);

    nng_close(pub);
    nng_close(sub);
    pthread_join(pub_thread, NULL);
    return 0;
}

# 增大文件描述符限制
ulimit -n 65536

# 增大 TCP 缓冲区
sysctl -w net.core.rmem_max=16777216
sysctl -w net.core.wmem_max=16777216
sysctl -w net.ipv4.tcp_rmem="4096 131072 16777216"
sysctl -w net.ipv4.tcp_wmem="4096 131072 16777216"

# 启用 TCP Fast Open
sysctl -w net.ipv4.tcp_fastopen=3

# 调整 TCP 拥塞算法
sysctl -w net.ipv4.tcp_congestion_control=bbr