nanomsg / NNG 消息库完全教程 / 第 5 章：NNG 现代 API 详解

#include <nng/nng.h>
#include <nng/protocol/pair0/pair.h>

int main() {
    nng_socket sock;
    int rv;

    // 创建 Socket
    if ((rv = nng_pair0_open(&sock)) != 0) {
        fprintf(stderr, "nng_pair0_open: %s\n", nng_strerror(rv));
        return 1;
    }

    // 监听
    if ((rv = nng_listen(sock, "tcp://*:5555", NULL, 0)) != 0) {
        fprintf(stderr, "nng_listen: %s\n", nng_strerror(rv));
        return 1;
    }

    // 使用 Socket ...
    nng_close(sock);
    return 0;
}

// listen: 监听传入连接（服务端模式）
nng_listener listener;
rv = nng_listen(sock, "tcp://*:5555", &listener, 0);

// dial: 主动连接（客户端模式）
nng_dialer dialer;
rv = nng_dial(sock, "tcp://server:5555", &dialer, 0);

nng_listener listener;
nng_socket sock;

nng_rep0_open(&sock);

// 创建 Listener（不立即启动）
nng_listener_create(&listener, sock);

// 配置 Listener 选项
nng_listener_setopt(listener, NNG_OPT_TCP_NODELAY, &(int){1}, sizeof(int));
nng_listener_setopt_string(listener, NNG_OPT_LISTEN_URL, "tcp://0.0.0.0:5555");

// 启动
nng_listener_start(listener, 0);

// 后续可关闭 Listener（不影响已建立的连接）
nng_listener_close(listener);

nng_dialer dialer;
nng_socket sock;

nng_req0_open(&sock);

// 创建 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);

// Dialer 会自动重连

// 发送
const char *msg = "Hello";
nng_send(sock, (void *)msg, strlen(msg), 0);

// 接收（需分配缓冲区）
char buf[256];
size_t sz = sizeof(buf);
nng_recv(sock, buf, &sz, 0);
printf("Received %zu bytes: %.*s\n", sz, (int)sz, buf);

void *buf = NULL;
size_t sz;
int rv = nng_recv(sock, &buf, &sz, NNG_FLAG_ALLOC);
if (rv == 0) {
    printf("Received: %.*s\n", (int)sz, buf);
    nng_free(buf, sz);  // 必须释放
}

nng_msg *msg;

// 分配消息
nng_msg_alloc(&msg, 0);

// 追加消息体
nng_msg_append(msg, "Hello", 5);

// 追加头部（协议头）
nng_msg_header(msg);  // 获取头部指针

// 发送消息对象
nng_sendmsg(sock, msg, 0);
// 注意：发送后 msg 被接管，不要再操作

// 接收消息对象
nng_msg *rmsg;
nng_recvmsg(sock, &rmsg, 0);

// 获取消息体
void *body = nng_msg_body(rmsg);
size_t len = nng_msg_len(rmsg);
printf("Body: %.*s\n", (int)len, (char *)body);

// 释放
nng_msg_free(rmsg);

nng_msg *msg;
nng_msg_alloc(&msg, 0);

// 在消息头部插入路由信息
nng_msg_insert(msg, "ROUTE", 5);

// 追加消息体
nng_msg_append(msg, "Payload", 7);

// 从头部裁剪（消费掉路由信息）
nng_msg_trim(msg, 5);

// 从尾部裁剪
nng_msg_chop(msg, 3);

#include <nng/nng.h>
#include <nng/protocol/reqrep0/rep.h>
#include <stdio.h>
#include <string.h>

// AIO 完成回调
void recv_callback(void *arg) {
    nng_aio *aio = (nng_aio *)arg;
    int rv = nng_aio_result(aio);

    if (rv != 0) {
        fprintf(stderr, "Async recv failed: %s\n", nng_strerror(rv));
        return;
    }

    // 获取接收到的消息
    nng_msg *msg = nng_aio_get_msg(aio);
    void *body = nng_msg_body(msg);
    size_t len = nng_msg_len(msg);
    printf("Async received: %.*s\n", (int)len, (char *)body);
    nng_msg_free(msg);
}

int main() {
    nng_socket sock;
    nng_aio *aio;
    int rv;

    nng_rep0_open(&sock);
    nng_listen(sock, "tcp://*:5555", NULL, 0);

    // 分配 AIO
    nng_aio_alloc(&aio, recv_callback, NULL);

    // 设置超时
    nng_aio_set_timeout(aio, 5000);  // 5 秒

    // 发起异步接收
    nng_recv_aio(sock, aio);

    // 主线程可以做其他事情...
    // nng_aio_wait(aio);  // 如果需要等待完成

    // 清理
    nng_aio_free(aio);
    nng_close(sock);
    return 0;
}

void send_callback(void *arg) {
    nng_aio *aio = (nng_aio *)arg;
    int rv = nng_aio_result(aio);
    if (rv == 0) {
        printf("Message sent successfully\n");
    } else {
        fprintf(stderr, "Send failed: %s\n", nng_strerror(rv));
    }
}

int main() {
    nng_socket sock;
    nng_aio *aio;

    nng_req0_open(&sock);
    nng_dial(sock, "tcp://server:5555", NULL, 0);

    nng_aio_alloc(&aio, send_callback, NULL);
    nng_aio_set_timeout(aio, 3000);

    // 准备消息
    nng_msg *msg;
    nng_msg_alloc(&msg, 0);
    nng_msg_append(msg, "Hello async!", 12);
    nng_aio_set_msg(aio, msg);

    // 异步发送
    nng_send_aio(sock, aio);

    nng_aio_wait(aio);
    nng_aio_free(aio);
    nng_close(sock);
    return 0;
}

#include <nng/nng.h>
#include <nng/protocol/reqrep0/rep.h>
#include <stdio.h>
#include <string.h>

typedef struct {
    nng_socket sock;
    nng_aio   *aio;
} server_ctx;

void server_recv_cb(void *arg);

void server_send_cb(void *arg) {
    server_ctx *ctx = (server_ctx *)arg;
    int rv = nng_aio_result(ctx->aio);
    if (rv != 0) {
        fprintf(stderr, "Send error: %s\n", nng_strerror(rv));
        return;
    }
    // 发送完成，继续接收下一个
    nng_recv_aio(ctx->sock, ctx->aio);
}

void server_recv_cb(void *arg) {
    server_ctx *ctx = (server_ctx *)arg;
    int rv = nng_aio_result(ctx->aio);

    if (rv != 0) {
        fprintf(stderr, "Recv error: %s\n", nng_strerror(rv));
        return;
    }

    // 处理消息
    nng_msg *msg = nng_aio_get_msg(ctx->aio);
    void *body = nng_msg_body(msg);
    size_t len = nng_msg_len(msg);
    printf("Received: %.*s\n", (int)len, (char *)body);

    // 构造响应（复用消息对象）
    nng_msg_clear(msg);
    nng_msg_append(msg, "OK", 2);
    nng_aio_set_msg(ctx->aio, msg);

    // 设置发送回调
    nng_aio_set_cb(ctx->aio, server_send_cb, ctx);
    nng_send_aio(ctx->sock, ctx->aio);
}

int main() {
    server_ctx ctx;
    int rv;

    if ((rv = nng_rep0_open(&ctx.sock)) != 0) {
        fprintf(stderr, "open: %s\n", nng_strerror(rv));
        return 1;
    }

    if ((rv = nng_listen(ctx.sock, "tcp://*:5555", NULL, 0)) != 0) {
        fprintf(stderr, "listen: %s\n", nng_strerror(rv));
        return 1;
    }

    nng_aio_alloc(&ctx.aio, server_recv_cb, &ctx);
    nng_aio_set_timeout(ctx.aio, 60000);

    // 启动异步接收循环
    nng_recv_aio(ctx.sock, ctx.aio);

    printf("Async REP server running on tcp://*:5555\n");

    // 主线程阻塞（或做其他工作）
    nng_aio_wait(ctx.aio);

    nng_aio_free(ctx.aio);
    nng_close(ctx.sock);
    return 0;
}

#include <nng/nng.h>
#include <nng/protocol/reqrep0/rep.h>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <pthread.h>

typedef struct {
    nng_ctx ctx;
    int     id;
} worker_args;

void *worker_thread(void *arg) {
    worker_args *wa = (worker_args *)arg;
    int id = wa->id;

    printf("Worker %d started\n", id);

    while (1) {
        nng_msg *msg;
        int rv;

        // 通过 Context 接收（独立的接收队列）
        rv = nng_ctx_recvmsg(wa->ctx, &msg, 0);
        if (rv != 0) {
            fprintf(stderr, "Worker %d recv error: %s\n", id, nng_strerror(rv));
            break;
        }

        void *body = nng_msg_body(msg);
        size_t len = nng_msg_len(msg);
        printf("Worker %d received: %.*s\n", id, (int)len, (char *)body);

        // 处理请求并准备响应
        nng_msg_clear(msg);
        char reply[64];
        snprintf(reply, sizeof(reply), "Response from worker %d", id);
        nng_msg_append(msg, reply, strlen(reply));

        // 通过 Context 发送
        rv = nng_ctx_sendmsg(wa->ctx, msg, 0);
        if (rv != 0) {
            fprintf(stderr, "Worker %d send error: %s\n", id, nng_strerror(rv));
            nng_msg_free(msg);
            break;
        }
    }

    return NULL;
}

int main() {
    nng_socket sock;
    int rv;
    int num_workers = 4;
    pthread_t threads[4];
    worker_args args[4];

    if ((rv = nng_rep0_open(&sock)) != 0) {
        fprintf(stderr, "nng_rep0_open: %s\n", nng_strerror(rv));
        return 1;
    }

    if ((rv = nng_listen(sock, "tcp://*:5555", NULL, 0)) != 0) {
        fprintf(stderr, "nng_listen: %s\n", nng_strerror(rv));
        return 1;
    }

    // 创建 Context 并启动工作线程
    for (int i = 0; i < num_workers; i++) {
        rv = nng_ctx_open(&args[i].ctx, sock);
        if (rv != 0) {
            fprintf(stderr, "nng_ctx_open: %s\n", nng_strerror(rv));
            return 1;
        }
        args[i].id = i;
        pthread_create(&threads[i], NULL, worker_thread, &args[i]);
    }

    printf("REP server with %d workers on tcp://*:5555\n", num_workers);

    // 等待所有线程
    for (int i = 0; i < num_workers; i++) {
        pthread_join(threads[i], NULL);
    }

    nng_close(sock);
    return 0;
}

// 设置接收超时 5 秒
nng_setopt_ms(sock, NNG_OPT_RECVTIMEO, 5000);

// 设置发送超时 3 秒
nng_setopt_ms(sock, NNG_OPT_SENDTIMEO, 3000);

// 设置最大接收消息 10MB
size_t maxsz = 10 * 1024 * 1024;
nng_setopt(sock, NNG_OPT_RECVMAXSZ, &maxsz, sizeof(maxsz));

// 设置 TCP NoDelay
int nodelay = 1;
nng_setopt(sock, NNG_OPT_TCP_NODELAY, &nodelay, sizeof(nodelay));

// 设置接收队列大小
int bufsize = 256;
nng_setopt(sock, NNG_OPT_RECVBUF, &bufsize, sizeof(bufsize));

// SUB 订阅
const char *topic = "weather";
nng_setopt(sock, NNG_OPT_SUB_SUBSCRIBE, topic, strlen(topic));

#define CHECK(rv, msg) do {         \
    if ((rv) != 0) {                \
        fprintf(stderr, "%s: %s\n", \
                msg, nng_strerror(rv)); \
        exit(1);                    \
    }                               \
} while (0)

int main() {
    nng_socket sock;
    int rv;

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

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

    // ... 使用 Socket ...

    nng_close(sock);
    return 0;
}

#include <nng/nng.h>
#include <nng/protocol/reqrep0/rep.h>
#include <stdio.h>
#include <string.h>
#include <signal.h>

static volatile int running = 1;

void handle_signal(int sig) {
    running = 0;
}

int main() {
    nng_socket sock;
    int rv;

    signal(SIGINT, handle_signal);
    signal(SIGTERM, handle_signal);

    // 创建 REP Socket
    if ((rv = nng_rep0_open(&sock)) != 0) {
        fprintf(stderr, "Socket: %s\n", nng_strerror(rv));
        return 1;
    }

    // 设置选项
    nng_setopt_ms(sock, NNG_OPT_RECVTIMEO, 1000);

    // 监听
    if ((rv = nng_listen(sock, "tcp://*:5555", NULL, 0)) != 0) {
        fprintf(stderr, "Listen: %s\n", nng_strerror(rv));
        nng_close(sock);
        return 1;
    }

    printf("NNG REP server running. Ctrl+C to stop.\n");

    while (running) {
        char *buf = NULL;
        size_t sz;

        rv = nng_recv(sock, &buf, &sz, NNG_FLAG_ALLOC);
        if (rv == NNG_ETIMEDOUT) {
            continue;  // 超时，检查 running 标志
        }
        if (rv != 0) {
            fprintf(stderr, "Recv: %s\n", nng_strerror(rv));
            break;
        }

        printf("Request: %.*s\n", (int)sz, buf);
        nng_free(buf, sz);

        const char *reply = "OK";
        if ((rv = nng_send(sock, (void *)reply, strlen(reply), 0)) != 0) {
            fprintf(stderr, "Send: %s\n", nng_strerror(rv));
            break;
        }
    }

    printf("\nShutting down...\n");
    nng_close(sock);
    return 0;
}