1. PUB/SUB 模式速览PUB/SUB 是 nanomsg 支持的 7 种消息模式之一对应NN_PUB/NN_SUB特点特性说明拓扑一对多扇出fan-outPUB 发一条 → 所有 SUB 都收到订阅过滤SUB 端可以设置前缀过滤器只接收感兴趣的话题无状态PUB 不关心谁在听发了就发了没有订阅者时消息直接丢弃单向流消息只能从 PUB → SUB反向发送会得到-ENOTSUP无 acceptServer 不需要显式accept()bind 后直接nn_send即可对比 REQ/REP 模式一问一答、有状态PUB/SUB 更适合日志分发、实时数据推送、行情广播等场景。2. 编译与运行2.1 编译确保已安装 nanomsg 开发库头文件 .so# 编译指令gcc-opubsub_demo pubsub_demo.c-lnanomsgildcmake..make2.2 运行Server 端先启动# 第 3 个参数 -s 表示 server 模式./pubsub_demo tcp://127.0.0.1:5555-sClient 端后启动可启动多个# 不带 -s 即 client 模式./pubsub_demo tcp://127.0.0.1:55552.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-s3. 程序架构一个 main两个角色intmain(intargc,char**argv){intrc;if((argc3)(strcmp(argv[2],-s)0)){rcserver(argv[1]);// Server: ./pubsub_demo url -s}elseif(argc2){rcclient(argv[1]);// Client: ./pubsub_demo url}else{fprintf(stderr,Usage: %s url [-s]\n,argv[0]);exit(EXIT_FAILURE);}exit(rc0?EXIT_SUCCESS:EXIT_FAILURE);}极简的参数分发-s走 Server否则走 Client。同一个可执行文件通过命令行决定角色——这在网络工具中非常常见想想curl既是下载工具也可以当代理。4. Server 端逐段精读4.1 创建 PUB Socketintserver(constchar*url){intfd;/* 创建 PUB socket。 AF_SP SP 协议族nanomsg 自定义的地址族 NN_PUB PUB/SUB 模式下的发布端 */fdnn_socket(AF_SP,NN_PUB);if(fd0){fprintf(stderr,nn_socket: %s\n,nn_strerror(nn_errno()));return(-1);}AF_SP是 nanomsg 的地址族常量类似AF_INET之于 BSD SocketNN_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, bind1) ├─ 解析 URL → prototcp, addr127.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);secshtonl(secs);// 主机字节序 → 网络字节序大端subshtonl(subs);/* 拼装二进制消息: [4字节 时间戳][4字节 订阅者数] 8 字节 */memcpy(msg,secs,sizeof(secs));memcpy(msgsizeof(secs),subs,sizeof(subs));/* 发送消息阻塞模式flags0 */rcnn_send(fd,msg,sizeof(msg),0);if(rc0){/* 发送失败时打印日志但不中断循环。 可能的原因: 没有订阅者、网络暂时不可用等。*/fprintf(stderr,nn_send: %s (ignoring)\n,nn_strerror(nn_errno()));}sleep(10);// 每 10 秒广播一次}几个值得注意的点消息是原始的字节流nanomsg 不关心消息内容是什么它只是把msg的 8 字节原样发给所有订阅者。序列化/反序列化完全由应用层决定。nn_get_statistic是实时统计NN_STAT_CURRENT_CONNECTIONS返回当前在线的订阅者连接数。nanomsg 在nn_sock_add连接建立时 1在nn_sock_rm连接断开时 -1。发送失败不中断当没有任何订阅者时nn_send会返回错误Connection refusedServer 打印一行日志后继续运行。这体现了 PUB 的即发即忘特性——没有听众也没关系。sleep(10)是简单的节奏控制生产环境可以用定时器或事件循环替代。5. Client 端逐段精读5.1 创建 SUB Socket 并连接intclient(constchar*url){intfd;intrc;/* 创建 SUB socket */fdnn_socket(AF_SP,NN_SUB);if(fd0){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 协议的订阅选项它工作的方式是按前缀匹配订阅前缀收到的消息(空)所有消息weather只收weather开头的消息sports.football只收sports.football开头的消息在我们的 demo 里因为消息是纯二进制不是文本且我们不关心过滤所以传空字符串接收全部。订阅是在 PUB 端执行的当 SUB 设置NN_SUB_SUBSCRIBE后这个订阅信息会被发送给 PUBPUB 端根据前缀决定是否把某条消息发给这个 SUB。这使得带宽浪费最小化——不需要的消息根本不会传输。5.3 接收循环for(;;){uint8_tmsg[2*sizeof(uint32_t)];// 8 字节缓冲区charhhmmss[9];// HH:MM:SS\0uint32_tsubs,secs;time_tt;/* 阻塞接收flags0 表示一直等到有消息 */rcnn_recv(fd,msg,sizeof(msg),0);if(rc0){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,msgsizeof(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);几点说明nn_recv是阻塞的没有消息时就阻塞等待有消息时返回实际收到的字节数。如果 socket 被关闭返回-EBADF。必须检查返回长度nn_recv返回的rc是实际收到的字节数。nanomsg 保证消息的原子性一条消息不会与下一条粘连但不保证你传的sizeof(msg)缓冲区总是够用——如果实际消息比你预料的长会截断且不会报错你需要用NN_RCVMAXSIZE选项来限制最大消息长度。htonl/ntohl对称使用Server 用htonl编码Client 用ntohl解码保证跨平台兼容。getpid()只是装饰打印 PID 可以区分是哪个 Client 进程的输出方便多 Client 测试时辨认。6. 消息格式与字节序6.1 消息布局0 4 8 (字节) ┌────────┬────────┐ │ secs │ subs │ │ uint32 │ uint32 │ └────────┴────────┘ 网络字节序大端这是一个非常简单的自定义二进制协议。没有 header、没有 magic number、没有 version——因为 demo 的目的就是展示最小可用程序。生产环境中应该至少加一个消息类型标识。6.2 为什么需要 htonl / ntohl不同 CPU 架构的整数存储方式不同架构字节序整数 0x12345678 的内存布局x86 / x86-64小端78 56 34 12网络字节序大端12 34 56 78ARM (大部分)小端78 56 34 12SPARC / PowerPC大端12 34 56 78如果 PUB 在 x86 上运行memcpy(msg, secs, 4)写入小端而 SUB 在 SPARC 上运行直接memcpy(secs, msg, 4)还原——得到的数字就全错了。htonlHost TO Network Long和ntohlNetwork 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 消息被截断设置最大接收消息长度intmax65536;nn_setsockopt(fd,NN_SOL_SOCKET,NN_RCVMAXSIZE,max,sizeof(max));nn_recv的缓冲区必须 ≥ 实际消息大小否则会被截断。7.3 如何实现非阻塞接收// 用 NN_DONTWAIT 标志rcnn_recv(fd,buf,sizeof(buf),NN_DONTWAIT);if(rc-EAGAIN){// 暂无消息做别的事}// 或者用 nn_poll 配合 poll/epollstructnn_pollfdpfd{.fdfd,.eventsNN_POLLIN};intnnn_poll(pfd,1,100);// 100ms 超时if(n0(pfd.reventsNN_POLLIN)){nn_recv(fd,buf,sizeof(buf),0);}8. 总结传输层由 URL 决定不需要在代码中显式选择——tcp://即 TCPipc://即 Unix 域套接字没有 accept——bind 后直接发connect 后直接收异步 IO 在框架内部处理NN_SUB_SUBSCRIBE是必选项——忘记设置会导致永久收不到消息二进制消息必须处理字节序——htonl编码、ntohl解码PUB 端即使没有订阅者也会继续运行——这是发布/订阅模式与 REQ/REP 的根本区别同一个可执行文件-s参数切换角色——这是一种整洁的程序设计模式