Nanomsg中pubsub_demo详解

1. PUB/SUB 模式速览

PUB/SUB 是 nanomsg 支持的 7 种消息模式之一（对应NN_PUB/NN_SUB），特点：

2. 编译与运行

2.1 编译

确保已安装 nanomsg 开发库（头文件 +.so）：

# 编译指令gcc-opubsub_demo pubsub_demo.c-lnanomsgildcmake..&&make

2.2 运行

Server 端（先启动）：

# 第 3 个参数 -s 表示 server 模式./pubsub_demo tcp://127.0.0.1:5555-s

Client 端（后启动，可启动多个）：

# 不带 -s 即 client 模式./pubsub_demo tcp://127.0.0.1:5555

2.3 预期输出

2.4 换用其他传输层

只需改 URL，代码不用动：

# 进程内通信（两个线程同一个进程，但这里 main 只支持单线程，仅作示例）./pubsub_demo inproc://mychannel-s# Unix 域套接字（同一台机器，比 TCP 更快）./pubsub_demo ipc:///tmp/demo.ipc-s# WebSocket./pubsub_demo ws://127.0.0.1:5555-s

3. 程序架构：一个 main，两个角色

intmain(intargc,char**argv){intrc;if((argc==3)&&(strcmp(argv[2],"-s")==0)){rc=server(argv[1]);// Server: ./pubsub_demo <url> -s}elseif(argc==2){rc=client(argv[1]);// Client: ./pubsub_demo <url>}else{fprintf(stderr,"Usage: %s <url> [-s]\n",argv[0]);exit(EXIT_FAILURE);}exit(rc==0?EXIT_SUCCESS:EXIT_FAILURE);}

极简的参数分发：-s走 Server，否则走 Client。同一个可执行文件，通过命令行决定角色——这在网络工具中非常常见（想想curl既是下载工具也可以当代理）。

4. Server 端逐段精读

4.1 创建 PUB Socket

intserver(constchar*url){intfd;/* 创建 PUB socket。 AF_SP = "SP 协议族"（nanomsg 自定义的地址族） NN_PUB = PUB/SUB 模式下的发布端 */fd=nn_socket(AF_SP,NN_PUB);if(fd<0){fprintf(stderr,"nn_socket: %s\n",nn_strerror(nn_errno()));return(-1);}

AF_SP是 nanomsg 的地址族常量（类似AF_INET之于 BSD Socket），NN_PUB指定了发布角色。nn_socket成功返回一个整数 fd（文件描述符），之后所有的操作都围绕这个 fd 进行。

注意：NN_PUB类型的 socket只能发送，如果调用nn_recv会返回-ENOTSUP。这是由socktype->flags中的NN_SOCKTYPE_FLAG_NORECV决定的，框架层面直接拒绝。

4.2 绑定地址

/* Bind 到指定 URL。这是同步操作——bind 成功后立即返回。 之后的客户端连接被 worker 线程异步处理， 调用者不需要显式 accept。 */if(nn_bind(fd,url)<0){fprintf(stderr,"nn_bind: %s\n",nn_strerror(nn_errno()));nn_close(fd);return(-1);}

nn_bind内部发生了什么：

nn_bind(fd, "tcp://127.0.0.1:5555") └─ nn_global_create_ep(sock, addr, bind=1) ├─ 解析 URL → proto="tcp", addr="127.0.0.1:5555" ├─ 匹配传输层 → NN_TCP ├─ nn_sock_add_ep(sock, tp, bind, addr) │ └─ nn_ep_init → tp->bind() │ └─ nn_btcp_create() ← TCP 传输层创建监听 socket └─ 返回 eid（端点 ID）

调用结束后，TCP 端口已经在监听，worker 线程在后台等待客户端连接。用户线程不需要做任何 accept 操作，直接去 publish 就行。

4.3 发布循环

for(;;){uint8_tmsg[2*sizeof(uint32_t)];// 消息缓冲区：2 个 uint32uint32_tsecs,subs;intrc;/* 准备消息内容 */secs=(uint32_t)time(NULL);// 当前 Unix 时间subs=(uint32_t)nn_get_statistic(fd,// 在线订阅者数NN_STAT_CURRENT_CONNECTIONS);secs=htonl(secs);// 主机字节序 → 网络字节序（大端）subs=htonl(subs);/* 拼装二进制消息: [4字节 时间戳][4字节 订阅者数] = 8 字节 */memcpy(msg,&secs,sizeof(secs));memcpy(msg+sizeof(secs),&subs,sizeof(subs));/* 发送消息（阻塞模式，flags=0） */rc=nn_send(fd,msg,sizeof(msg),0);if(rc<0){/* 发送失败时打印日志但不中断循环。 可能的原因: 没有订阅者、网络暂时不可用等。*/fprintf(stderr,"nn_send: %s (ignoring)\n",nn_strerror(nn_errno()));}sleep(10);// 每 10 秒广播一次}

几个值得注意的点：

1. 消息是原始的字节流：nanomsg 不关心消息内容是什么，它只是把msg的 8 字节原样发给所有订阅者。序列化/反序列化完全由应用层决定。

2. nn_get_statistic是实时统计：NN_STAT_CURRENT_CONNECTIONS返回当前在线的订阅者连接数。nanomsg 在nn_sock_add（连接建立）时 +1，在nn_sock_rm（连接断开）时 -1。

3. 发送失败不中断：当没有任何订阅者时，nn_send会返回错误（Connection refused），Server 打印一行日志后继续运行。这体现了 PUB 的"即发即忘"特性——没有听众也没关系。

4. sleep(10)是简单的节奏控制：生产环境可以用定时器或事件循环替代。

5. Client 端逐段精读

5.1 创建 SUB Socket 并连接

intclient(constchar*url){intfd;intrc;/* 创建 SUB socket */fd=nn_socket(AF_SP,NN_SUB);if(fd<0){fprintf(stderr,"nn_socket: %s\n",nn_strerror(nn_errno()));return(-1);}/* 主动连接到 PUB server */if(nn_connect(fd,url)<0){fprintf(stderr,"nn_socket: %s\n",nn_strerror(nn_errno()));nn_close(fd);return(-1);}

NN_SUB类型的 socket只能接收（NN_SOCKTYPE_FLAG_NOSEND），调用nn_send会返回-ENOTSUP。nn_connect是异步的——函数返回时连接可能还没建立好，但 nanomsg 会自动在后台重试。

5.2 设置订阅过滤器——极其重要！

/* 订阅所有消息：前缀为空字符串 ""。 这是 NN_SUB 关键的一步——不订阅就收不到任何消息！ */if(nn_setsockopt(fd,NN_SUB,NN_SUB_SUBSCRIBE,"",0)<0){fprintf(stderr,"nn_setsockopt: %s\n",nn_strerror(nn_errno()));nn_close(fd);return(-1);}

这是整个程序中最容易被忽略但最致命的一行。没有这行，Client 一条消息都收不到！

NN_SUB_SUBSCRIBE是 SUB 协议的订阅选项，它工作的方式是按前缀匹配：

在我们的 demo 里，因为消息是纯二进制（不是文本），且我们不关心过滤，所以传空字符串接收全部。

订阅是在 PUB 端执行的！当 SUB 设置NN_SUB_SUBSCRIBE后，这个订阅信息会被发送给 PUB，PUB 端根据前缀决定是否把某条消息发给这个 SUB。这使得带宽浪费最小化——不需要的消息根本不会传输。

5.3 接收循环

for(;;){uint8_tmsg[2*sizeof(uint32_t)];// 8 字节缓冲区charhhmmss[9];// "HH:MM:SS\0"uint32_tsubs,secs;time_tt;/* 阻塞接收，flags=0 表示一直等到有消息 */rc=nn_recv(fd,msg,sizeof(msg),0);if(rc<0){fprintf(stderr,"nn_recv: %s\n",nn_strerror(nn_errno()));break;// 出错 → 退出循环}/* 长度检查：nanomsg 不保证分包，所以必须验证 */if(rc!=sizeof(msg)){fprintf(stderr,"nn_recv: got %d bytes, wanted %d\n",rc,(int)sizeof(msg));break;}/* 从二进制消息中还原两个 uint32 */memcpy(&secs,msg,sizeof(secs));memcpy(&subs,msg+sizeof(secs),sizeof(subs));/* 网络字节序 → 主机字节序，转换为本地时间并格式化 */t=(time_t)ntohl(secs);strftime(hhmmss,sizeof(hhmmss),"%T",localtime(&t));/* 打印：时间 + 进程 ID + 在线订阅者数 */printf("%s <pid %u> There are %u clients connected.\n",hhmmss,(unsigned)getpid(),(unsigned)ntohl(subs));}nn_close(fd);return(-1);

几点说明：

1. nn_recv是阻塞的：没有消息时就阻塞等待，有消息时返回实际收到的字节数。如果 socket 被关闭，返回-EBADF。

2. 必须检查返回长度：nn_recv返回的rc是实际收到的字节数。nanomsg 保证消息的原子性（一条消息不会与下一条粘连），但不保证你传的sizeof(msg)缓冲区总是够用——如果实际消息比你预料的长，会截断且不会报错（你需要用NN_RCVMAXSIZE选项来限制最大消息长度）。

3. htonl/ntohl对称使用：Server 用htonl编码，Client 用ntohl解码，保证跨平台兼容。

4. getpid()只是装饰：打印 PID 可以区分是哪个 Client 进程的输出，方便多 Client 测试时辨认。

6. 消息格式与字节序

6.1 消息布局

0 4 8 (字节) ┌────────┬────────┐ │ secs │ subs │ │ uint32 │ uint32 │ └────────┴────────┘ 网络字节序（大端）

这是一个非常简单的自定义二进制协议。没有 header、没有 magic number、没有 version——因为 demo 的目的就是展示最小可用程序。生产环境中应该至少加一个消息类型标识。

6.2 为什么需要 htonl / ntohl？

不同 CPU 架构的整数存储方式不同：

如果 PUB 在 x86 上运行memcpy(msg, &secs, 4)（写入小端），而 SUB 在 SPARC 上运行直接memcpy(&secs, msg, 4)还原——得到的数字就全错了。

htonl（Host TO Network Long）和ntohl（Network TO Host Long）确保：

• x86 发送时：小端 → 大端（htonl执行字节反转）
• SPARC 接收时：大端 → 大端（ntohl不执行任何操作）
• 两端一致解码出正确的数值

关键规则：发送方一律htonl，接收方一律ntohl。即使在两台都是 x86 的机器上通信，这个习惯也不能省——代码就应该写成平台无关的。

7. 常见问题与调优

7.1 如何做话题过滤？

// 只订阅 "weather" 开头的消息nn_setsockopt(fd,NN_SUB,NN_SUB_SUBSCRIBE,"weather",7);// 可以多次调用，订阅多个前缀nn_setsockopt(fd,NN_SUB,NN_SUB_SUBSCRIBE,"sports",6);nn_setsockopt(fd,NN_SUB,NN_SUB_SUBSCRIBE,"finance",7);

Server 发送时在消息头加话题前缀：

nn_send(fd,"weather: Beijing 25°C",21,0);nn_send(fd,"sports: Football 2:1",20,0);

7.2 消息被截断？

设置最大接收消息长度：

intmax=65536;nn_setsockopt(fd,NN_SOL_SOCKET,NN_RCVMAXSIZE,&max,sizeof(max));

nn_recv的缓冲区必须 ≥ 实际消息大小，否则会被截断。

7.3 如何实现非阻塞接收？

// 用 NN_DONTWAIT 标志rc=nn_recv(fd,buf,sizeof(buf),NN_DONTWAIT);if(rc==-EAGAIN){// 暂无消息，做别的事}// 或者用 nn_poll 配合 poll/epollstructnn_pollfdpfd={.fd=fd,.events=NN_POLLIN};intn=nn_poll(&pfd,1,100);// 100ms 超时if(n>0&&(pfd.revents&NN_POLLIN)){nn_recv(fd,buf,sizeof(buf),0);}

8. 总结

1. 传输层由 URL 决定，不需要在代码中显式选择——tcp://即 TCP，ipc://即 Unix 域套接字
2. 没有 accept——bind 后直接发，connect 后直接收，异步 IO 在框架内部处理
3. NN_SUB_SUBSCRIBE是必选项——忘记设置会导致永久收不到消息
4. 二进制消息必须处理字节序——htonl编码、ntohl解码
5. PUB 端即使没有订阅者也会继续运行——这是发布/订阅模式与 REQ/REP 的根本区别
6. 同一个可执行文件，-s参数切换角色——这是一种整洁的程序设计模式