nanomsg / NNG 消息库完全教程 / 第 8 章：IPC 与进程间通信

#include <nng/nng.h>
#include <nng/protocol/pubsub0/pub.h>
#include <nng/protocol/pubsub0/sub.h>
#include <stdio.h>
#include <string.h>
#include <pthread.h>
#include <unistd.h>

void *subscriber_thread(void *arg) {
    nng_socket sub;
    nng_sub0_open(&sub);

    // 连接到 inproc 地址
    nng_setopt(sub, NNG_OPT_SUB_SUBSCRIBE, "", 0);
    nng_dial(sub, "inproc://events", NULL, 0);

    printf("Subscriber thread started\n");

    for (int i = 0; i < 5; i++) {
        char *buf = NULL;
        size_t sz;
        if (nng_recv(sub, &buf, &sz, NNG_FLAG_ALLOC) == 0) {
            printf("[SUB] Received: %.*s\n", (int)sz, buf);
            nng_free(buf, sz);
        }
    }

    nng_close(sub);
    return NULL;
}

int main() {
    nng_socket pub;
    pthread_t sub_tid;

    nng_pub0_open(&pub);

    // 绑定到 inproc 地址
    nng_listen(pub, "inproc://events", NULL, 0);

    // 启动订阅者线程
    pthread_create(&sub_tid, NULL, subscriber_thread, NULL);
    usleep(100000);  // 等待订阅者连接

    // 发布消息
    const char *events[] = {
        "user.login",
        "order.created",
        "payment.success",
        "order.shipped",
        "user.logout"
    };

    for (int i = 0; i < 5; i++) {
        printf("[PUB] Publishing: %s\n", events[i]);
        nng_send(pub, (void *)events[i], strlen(events[i]), 0);
        usleep(200000);
    }

    pthread_join(sub_tid, NULL);
    nng_close(pub);

    printf("Done.\n");
    return 0;
}

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

#include <nng/nng.h>
#include <nng/protocol/pair0/pair.h>
#include <stdio.h>
#include <string.h>
#include <pthread.h>
#include <unistd.h>

typedef struct {
    nng_socket sock;
    const char *name;
} thread_arg;

void *worker(void *arg) {
    thread_arg *ta = (thread_arg *)arg;

    nng_pair0_open(&ta->sock);
    nng_dial(ta->sock, "inproc://control", NULL, 0);

    // 发送就绪消息
    nng_send(ta->sock, "READY", 5, 0);

    // 接收工作指令
    char *buf = NULL;
    size_t sz;
    if (nng_recv(ta->sock, &buf, &sz, NNG_FLAG_ALLOC) == 0) {
        printf("[%s] Got command: %.*s\n", ta->name, (int)sz, buf);
        nng_free(buf, sz);
    }

    // 发送结果
    nng_send(ta->sock, "DONE", 4, 0);

    nng_close(ta->sock);
    return NULL;
}

int main() {
    nng_socket pair;
    pthread_t tid;
    thread_arg arg = { .name = "Worker-1" };

    nng_pair0_open(&pair);
    nng_listen(pair, "inproc://control", NULL, 0);

    pthread_create(&tid, NULL, worker, &arg);
    usleep(100000);

    // 接收就绪消息
    char *buf = NULL;
    size_t sz;
    nng_recv(pair, &buf, &sz, NNG_FLAG_ALLOC);
    printf("[Main] Worker status: %.*s\n", (int)sz, buf);
    nng_free(buf, sz);

    // 发送工作指令
    nng_send(pair, "PROCESS_DATA", 12, 0);

    // 接收结果
    nng_recv(pair, &buf, &sz, NNG_FLAG_ALLOC);
    printf("[Main] Worker result: %.*s\n", (int)sz, buf);
    nng_free(buf, sz);

    pthread_join(tid, NULL);
    nng_close(pair);
    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 on_signal(int sig) { running = 0; }

int main() {
    nng_socket sock;
    int rv;

    signal(SIGINT, on_signal);
    signal(SIGTERM, on_signal);

    nng_rep0_open(&sock);

    // 设置最大接收消息大小
    size_t maxsz = 1048576;  // 1MB
    nng_setopt(sock, NNG_OPT_RECVMAXSZ, &maxsz, sizeof(maxsz));

    // 设置接收超时
    nng_setopt_ms(sock, NNG_OPT_RECVTIMEO, 1000);

    // 绑定到 Unix Socket
    if ((rv = nng_listen(sock, "ipc:///tmp/nng_demo.ipc", NULL, 0)) != 0) {
        fprintf(stderr, "nng_listen: %s\n", nng_strerror(rv));
        return 1;
    }

    printf("IPC server listening on /tmp/nng_demo.ipc\n");

    while (running) {
        char *buf = NULL;
        size_t sz;
        rv = nng_recv(sock, &buf, &sz, NNG_FLAG_ALLOC);
        if (rv == NNG_ETIMEDOUT) continue;
        if (rv != 0) break;

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

        const char *reply = "IPC OK";
        nng_send(sock, (void *)reply, strlen(reply), 0);
    }

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

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

int main() {
    nng_socket sock;
    int rv;

    nng_req0_open(&sock);

    // 连接到 Unix Socket
    if ((rv = nng_dial(sock, "ipc:///tmp/nng_demo.ipc", NULL, 0)) != 0) {
        fprintf(stderr, "nng_dial: %s\n", nng_strerror(rv));
        return 1;
    }

    // 发送请求
    const char *msg = "Hello via IPC!";
    nng_send(sock, (void *)msg, strlen(msg), 0);

    // 接收响应
    char *buf = NULL;
    size_t sz;
    if (nng_recv(sock, &buf, &sz, NNG_FLAG_ALLOC) == 0) {
        printf("Reply: %.*s\n", (int)sz, buf);
        nng_free(buf, sz);
    }

    nng_close(sock);
    return 0;
}

# 终端 1
cc ipc_server.c -lnng -o ipc_server && ./ipc_server

# 终端 2
cc ipc_client.c -lnng -o ipc_client && ./ipc_client
# 输出: Reply: IPC OK

// 在绑定后设置 Socket 文件权限
#include <sys/stat.h>
chmod("/tmp/nng_demo.ipc", 0660);  // 仅所有者和组可访问

# /etc/systemd/system/myapp.socket
[Unit]
Description=My NNG IPC Socket

[Socket]
ListenStream=/var/run/myapp.sock
SocketMode=0660

[Install]
WantedBy=sockets.target

# 查找残留的 IPC Socket 文件
ls -la /tmp/*.ipc /var/run/*.sock

# 启动前清理
rm -f /tmp/nng_demo.ipc

# 在代码中处理
#include <unistd.h>
unlink("/tmp/nng_demo.ipc");  // 启动前删除旧文件
nng_listen(sock, "ipc:///tmp/nng_demo.ipc", NULL, 0);