nanomsg / NNG 消息库完全教程 / 第 10 章：最佳实践与总结

#include <nng/nng.h>
#include <stdio.h>
#include <stdlib.h>

// 致命错误：打印并退出
#define FATAL(fmt, ...) do {                    \
    fprintf(stderr, "[FATAL] " fmt "\n",        \
            ##__VA_ARGS__);                      \
    exit(1);                                    \
} while (0)

// 非致命错误：打印并继续
#define WARN(fmt, ...) do {                     \
    fprintf(stderr, "[WARN] " fmt "\n",         \
            ##__VA_ARGS__);                      \
} while (0)

// 检查 NNG 返回值
#define NNG_CHECK(rv, msg) do {                 \
    if ((rv) != 0) {                            \
        FATAL("%s: %s (code=%d)",               \
              msg, nng_strerror(rv), rv);        \
    }                                           \
} while (0)

// 使用示例
int main() {
    nng_socket sock;
    int rv;

    rv = nng_req0_open(&sock);
    NNG_CHECK(rv, "nng_req0_open");

    rv = nng_dial(sock, "tcp://localhost:5555", NULL, 0);
    NNG_CHECK(rv, "nng_dial");

    nng_close(sock);
    return 0;
}

#include <nng/nng.h>
#include <stdio.h>
#include <unistd.h>

#define MAX_RETRIES 5
#define RETRY_DELAY_MS 1000

int send_with_retry(nng_socket sock, const void *data, size_t len) {
    int rv;
    for (int attempt = 1; attempt <= MAX_RETRIES; attempt++) {
        rv = nng_send(sock, (void *)data, len, 0);
        if (rv == 0) {
            return 0;  // 成功
        }

        WARN("send attempt %d/%d failed: %s",
             attempt, MAX_RETRIES, nng_strerror(rv));

        if (rv == NNG_ETIMEDOUT || rv == NNG_EAGAIN) {
            // 可重试的错误
            usleep(RETRY_DELAY_MS * 1000 * attempt);  // 指数退避
        } else {
            // 不可重试的错误
            break;
        }
    }
    return rv;
}

int recv_with_retry(nng_socket sock, void **buf, size_t *sz) {
    int rv;
    for (int attempt = 1; attempt <= MAX_RETRIES; attempt++) {
        rv = nng_recv(sock, buf, sz, NNG_FLAG_ALLOC);
        if (rv == 0) {
            return 0;
        }

        if (rv == NNG_ETIMEDOUT) {
            WARN("recv timeout, attempt %d/%d", attempt, MAX_RETRIES);
            continue;
        }

        // 其他错误不重试
        WARN("recv failed: %s", nng_strerror(rv));
        break;
    }
    return rv;
}

#include <nng/nng.h>
#include <signal.h>
#include <stdio.h>
#include <stdbool.h>

static volatile bool g_running = true;

void signal_handler(int sig) {
    (void)sig;
    g_running = false;
}

int main() {
    nng_socket sock;
    int rv;

    // 注册信号处理
    signal(SIGINT, signal_handler);
    signal(SIGTERM, signal_handler);

    // 设置 linger 时间（等待未发送消息的时间）
    int linger_ms = 5000;
    nng_setopt(sock, NNG_OPT_LINGER, &linger_ms, sizeof(linger_ms));

    // 设置接收超时（让循环能检查 g_running）
    nng_setopt_ms(sock, NNG_OPT_RECVTIMEO, 1000);

    // ... 初始化 socket ...

    while (g_running) {
        // 收发消息（超时后会返回，检查 g_running）
        char *buf = NULL;
        size_t sz;
        rv = nng_recv(sock, &buf, &sz, NNG_FLAG_ALLOC);
        if (rv == NNG_ETIMEDOUT) {
            continue;  // 检查 g_running
        }
        if (rv != 0) {
            WARN("recv error: %s", nng_strerror(rv));
            break;
        }

        // 处理消息...
        nng_free(buf, sz);
    }

    // 优雅关闭
    printf("Shutting down...\n");
    nng_close(sock);  // 等待 linger 时间
    printf("Done.\n");
    return 0;
}

// ❌ 错误：频繁发送小消息
for (int i = 0; i < 1000; i++) {
    nng_send(sock, &data[i], 1, 0);  // 1 字节 × 1000 次
}

// ✅ 正确：批量发送
nng_send(sock, data, 1000, 0);  // 1000 字节 × 1 次

// ❌ 错误：每次发送都分配/释放内存
while (running) {
    char *msg = malloc(1024);
    fill_data(msg, 1024);
    nng_send(sock, msg, 1024, 0);
    free(msg);  // 发送后立即释放（但 send 可能还在进行）
}

// ✅ 正确：使用零拷贝或预分配
char msg_buf[1024];  // 预分配
while (running) {
    fill_data(msg_buf, 1024);
    nng_send(sock, msg_buf, 1024, 0);  // NNG 内部会拷贝
}

// ✅ 正确：服务端使用 listen，客户端使用 dial
// 服务端
nng_listen(sock, "tcp://0.0.0.0:5555", NULL, 0);

// 客户端（支持自动重连）
nng_dialer dialer;
nng_dialer_create(&dialer, sock);
nng_dialer_setopt_ms(dialer, NNG_OPT_RECONNMINT, 100);
nng_dialer_setopt_ms(dialer, NNG_OPT_RECONNMAXT, 30000);
nng_dialer_start(dialer, 0);

// ❌ 错误：REQ 连续发送两次
nng_send(sock, "msg1", 4, 0);
nng_send(sock, "msg2", 4, 0);  // NNG_ESTATE! REQ 必须先收后发

// ✅ 正确：REQ 严格交替 send-recv
nng_send(sock, "msg1", 4, 0);
nng_recv(sock, buf, &sz, 0);
nng_send(sock, "msg2", 4, 0);
nng_recv(sock, buf, &sz, 0);

// ❌ 错误：SUB 没有设置订阅
nng_sub0_open(&sock);
nng_dial(sock, "tcp://localhost:5556", NULL, 0);
nng_recv(sock, buf, &sz, 0);  // 永远收不到消息！

// ✅ 正确：设置订阅过滤
nng_setopt(sock, NNG_OPT_SUB_SUBSCRIBE, "", 0);  // 接收所有

// ❌ 错误：零拷贝接收后忘记释放
void *buf = NULL;
nng_recv(sock, &buf, &sz, NNG_FLAG_ALLOC);
// ... 使用 buf ...
// 忘记 nng_free(buf, sz); ← 内存泄漏！

// ✅ 正确：始终释放
nng_free(buf, sz);

// ❌ 错误：sendmsg 后再次使用消息
nng_msg *msg;
nng_msg_alloc(&msg, 0);
nng_msg_append(msg, "data", 4);
nng_sendmsg(sock, msg, 0);
nng_msg_free(msg);  // 错误！sendmsg 已接管所有权

// ✅ 正确：sendmsg 后不要释放
nng_sendmsg(sock, msg, 0);
// msg 已被 NNG 接管，不要操作

// ❌ 错误：inproc 先 connect 后 listen
nng_dial(sock, "inproc://test", NULL, 0);  // 可能失败
nng_listen(other, "inproc://test", NULL, 0);

// ✅ 正确：先 listen 后 connect
nng_listen(other, "inproc://test", NULL, 0);
nng_dial(sock, "inproc://test", NULL, 0);

// 建议收集的指标
typedef struct {
    uint64_t messages_sent;
    uint64_t messages_received;
    uint64_t send_errors;
    uint64_t recv_errors;
    uint64_t reconnect_count;
    double   avg_latency_us;
    double   p99_latency_us;
} nng_metrics;

// 使用原子操作更新
#include <stdatomic.h>

static atomic_uint_fast64_t g_sent = 0;
static atomic_uint_fast64_t g_recv = 0;

// 在发送时
atomic_fetch_add(&g_sent, 1);

// 在接收时
atomic_fetch_add(&g_recv, 1);