1. 先把ZMQ是什么说清楚消息队列之外的另一种选择第一次接触ZeroMQ是在做一个跨平台的数据分发模块当时项目里用的是传统的TCP长连接服务端维护几百个客户端连接每个连接单独开线程收发一遇到客户端闪断重连就要处理一堆半包粘包的问题代码越写越厚稳定性还不见得好。后来在某个技术方案评审里看到有人提ZeroMQ最开始我以为它就是又一个消息队列直到真的上手才发现它跟MQ完全是两种思维。ZeroMQ常被简称为ZMQ的本质是一个网络通信库不是消息中间件。它没有独立的Broker进程不需要单独部署服务器它把你的应用程序直接变成了通信节点。你在代码里创建一个socket这个socket就能像传统socket一样收发数据但它内部封装了连接管理、断线重连、消息分帧、排队缓冲这些脏活累活。也就是说ZMQ给开发者的核心价值是你只需要关心业务消息本身不需要关心底层连接怎么建立、断了怎么处理、消息怎么不丢。引用官网上那句著名的话ZMQ是激进的socket它让socket变成了一个高性能消息队列的载体。这句话怎么理解传统socket是端点你要自己管理连接生命周期ZMQ的socket是通信模式的最小单元它自带策略这些策略包括自动重连、队列缓存、多传输协议支持、线程安全等等。换句话说你创建一个ZMQ的REQ socket它天然就带有请求-应答这种交互模式的所有语义你再也不用自己在应用层实现等回复超时重发这类逻辑。这篇文章先讲ZMQ最核心的三类通信模式请求-应答REQ-REP、发布-订阅PUB-SUB、推拉PUSH-PULL。之所以分成上和下因为ZMQ的模式体系其实不止这三类还有DEALER-ROUTER这类高级路由模式、PAIR双工模式以及在通信模式之上衍生的各种扩展能力这些内容留到下篇。把基础三模式吃透你基本能解决工作中八成以上的进程通信需求。先说清楚一个基本定位ZMQ适合什么样的读者适合正在做分布式系统、微服务之间通信、数据采集管道、实时数据推送这类工作的开发者。如果你只是要在两台机器之间传个文件那直接用HTTP就行没必要引入ZMQ但如果你面对的是几十上百个节点之间互相通信并且对延迟敏感、对连接可靠性要求高那ZMQ的性价比就非常明显了。2. 三种通信模式的基础认知与适用边界ZMQ把通信模式封装成了不同Socket类型每种Socket类型代表一种确定的通信语义。这一点跟传统Socket完全不同传统Socket只定义了TCP/UDP这类传输层语义而ZMQ的Socket直接定义到了业务交互层。理解这一点是理解ZMQ一切的起点。2.1 REQ-REP一问一答的同步远程调用REQ-REP是最容易上手的一种模式它的交互模型跟HTTP请求-响应非常像客户端发一个请求服务端收一个请求回一个应答客户端收到应答之后才能发下一个请求。这种严格的一来一回让它在RPC场景中非常讨喜。打个比方REQ-REP就像你去窗口办事你递材料请求窗口给你办完递回执应答你才能办下一件事。你不能一次递两份材料也不能不拿回执就递下一份这是协议规定死的流程。这个模式解决的核心问题是远程调用的同步语义。当你的系统里有一个计算服务其他模块需要调用它调用方希望我发一个任务必须等到结果返回那REQ-REP就是最合适的骨架。2.2 PUB-SUB一对多的实时数据广播PUB-SUB的模型是从新闻订阅演化来的发布者PUB只管把消息发出去不关心谁在听、听没听到订阅者SUB可以动态地加入和离开只接收自己感兴趣的消息。这个模式的最大特点是解耦——发布者完全不知道订阅者的存在两者之间没有直接的联系。打个比方PUB-SUB就像广播电台电台播新闻你随时可以拧开收音机听你也可以随时关掉。电台不知道有多少人在听也不会因为你关掉收音机就不播了。这个模式解决的核心问题是一对多数据分发。当你的系统里有一个数据源需要同时推送给多个消费方并且消费方的数量会动态变化那PUB-SUB就是最自然的选择。有一个关键特征要特别注意订阅者只能收到它上线之后发布的消息早于订阅时间的消息不会补发。这一点很像现实中的广播——你打开收音机之前播过的内容你不可能听到。2.3 PUSH-PULL数据流水线与负载均衡PUSH-PULL是三模式里最工业流水线的一个。PUSH端把任务分发给一组PULL端PULL端处理完还可以再PUSH给下游形成一条数据流水线。最经典的用法是MapReduce风格的任务分发一个PUSH端把大量小任务均匀分发给多个Worker每个Worker处理完自己那份任务再汇总到一个PULL端收集结果。打个比方PUSH-PULL就像一条工厂流水线上游把零件放到传送带工位上的工人轮流拿零件加工加工完放到下一条传送带。上游不关心具体是哪个工人拿了零件只管把零件放上传送带就行。这个模式解决的核心问题是并行任务分发与结果汇聚。当你的系统里有一个大任务可以拆成很多小任务并且要并行处理那PUSH-PULL就是标准答案。3. REQ-REP模式从理论到落地REQ-REP在ZMQ里对应的Socket类型是REQ和REP。先说一个代码骨架然后再做详细解读。这里用Python的pyzmq库做演示它是官方支持的语言绑定中语法最直观的一个。3.1 最基本的REQ-REP代码结构服务端代码import zmq context zmq.Context() socket context.socket(zmq.REP) socket.bind(tcp://*:5555) while True: # 必须先接收请求 request socket.recv() print(f收到请求: {request.decode()}) # 处理后必须发送应答 reply freply for {request.decode()}.encode() socket.send(reply)客户端代码import zmq context zmq.Context() socket context.socket(zmq.REQ) socket.connect(tcp://localhost:5555) # 必须先发送请求 socket.send(bhello) # 再等待应答 reply socket.recv() print(f收到应答: {reply.decode()})这两段代码量不大但里面包含了REQ-REP模式最核心的纪律REQ端必须严格走发送-接收-发送-接收的交替流程REP端必须严格走接收-发送-接收-发送的交替流程。任何一步乱了ZMQ会直接抛异常这在设计上是一个强制约束。为什么要强行要求这种交替因为ZMQ内部为REQ和REP做了状态机管理。REQ socket在发送完请求后状态从可发送变为等待应答只有等收到应答状态才会回到可发送。这个设计避免了传统Socket里常见的客户端发两次请求但只收到一次应答这类时序错乱问题。运行上面的代码先启动服务端再启动客户端终端输出各自打印一行消息。整个过程不需要任何额外配置两端连上就能通信这也是ZMQ其中一个优点基础使用成本极低。3.2 REQ-REP生产级代码的调整方向上面的骨架代码只适合验证连通性真实生产环境必须做三处调整。第一send和recv要用带超时的方式。默认情况下ZMQ的recv是阻塞的如果服务端挂在某个请求上客户端就永远等下去。给socket设置RCVTIMEO和SNDTIMEO把阻塞变成有阈值的等待。在pyzmq里可以这样socket.setsockopt(zmq.RCVTIMEO, 5000) # 接收超时5秒 socket.setsockopt(zmq.SNDTIMEO, 5000) # 发送超时5秒设置超时需要配合捕获异常用try except通常是zmq.error.Again。当超时触发时说明对端出现过载或死锁此时合理的做法是丢弃当前请求上下文重新进入下一个接收周期。第二周期性心跳。长时间运行的服务端如果客户端崩了TCP长连接在底层可能需要很长时间才能感知到断线。建议在业务间隙发心跳包或者利用ZMQ的ZMQ_HEARTBEAT_IVL选项设置心跳间隔。心跳的意义不是保证消息一定送达而是缩短断线探测时间让Socket在底层尽早重建连接。第三高水位标记High Water MarkHWM。REQ-REP模式中如果服务端处理不过来消息会在两端的内存队列里堆积。HWM就是给这个队列设置一个上限值。发送端超过HWM时会进入阻塞这实际上是一种背压机制防止内存被无限积压的消息占满。在pyzmq里这样设置socket.setsockopt(zmq.SNDHWM, 1000) # 发送队列最多1000条 socket.setsockopt(zmq.RCVHWM, 1000) # 接收队列最多1000条注意HWM的默认值是1000但在实际项目中我建议你根据单条消息的大小重新评估。如果单条消息是100KB1000条就是100MB这个缓冲规模对于长期运行的服务端来说很危险。合理做法是估算单条消息大小和可接受的最大积压反推HWM数值。3.3 REQ-REP的常见失败模式与应对REQ-REP虽然简单但有三类问题在实践里经常出现。第一类REQ端连续多次发送不接收。这是新手最常犯的错误代码里连续写两个send第二个send会直接报错operation cannot be accomplished in current state。解决办法是严格遵守交替流程如果确有批量请求要发不要用同一个REQ socket顺序发改用DEALER socket。这也是ZMQ后续版本把DEALER作为REQ的异步亲戚引入的原因。第二类REP端一收一发的顺序颠倒。新手有时会忍不住在recv之前先send一个消息比如主动回一个服务端已就绪的通知这在REP socket上直接被禁止。如果你需要这种推语义请换用PUSH或PUB模式不要试图在REP上实现。第三类服务端多线程处理请求时的Socket共享问题。ZMQ的Socket默认不是线程安全的。多个线程共用一个REP socket会出现状态混乱消息串线程。正确做法是一线程一Socket或者用ZMQ的inproc协议配合代理模式做转发。在实践中我会在服务端进程里创建一个独立的轮询线程专门负责recv然后把业务处理交给线程池处理完再通过该线程调send这样就锁定了Socket只在单个线程内使用。4. PUB-SUB模式的消息分发与丢失问题PUB-SUB在ZMQ的三种基础模式里最受关注因为它直接触及了一个核心痛点怎么样才能让一个数据源高效地把数据发给大量消费者。但PUB-SUB也是踩坑最多的一个模式因为它有一个与生俱来的数据特性先订阅后接收提前订阅也不保证不丢。4.1 PUB-SUB代码骨架与关键参数发布端代码import zmq import time context zmq.Context() socket context.socket(zmq.PUB) socket.bind(tcp://*:5556) while True: message fmsg-{time.time()} socket.send(message.encode()) time.sleep(1)订阅端代码import zmq context zmq.Context() socket context.socket(zmq.SUB) socket.connect(tcp://localhost:5556) # 关键必须设置订阅过滤器 socket.setsockopt_string(zmq.SUBSCRIBE, ) while True: message socket.recv() print(message.decode())这段代码里有一行是PUB-SUB模式的关键SUBSCRIBE过滤器。如果订阅端不调用setsockopt_string(zmq.SUBSCRIBE, ...)它什么都收不到。空字符串表示接收所有前缀为空的Topic——实际上就是接收所有消息。这种设计是为了性能考虑消息过滤在订阅端做。发布端把消息广播给所有订阅者每个订阅者收到后在本地按Topic前缀做过滤。这个机制的优点是发布端完全不需要知道订阅者关心什么缺点是要把消息先发到网络上走一圈过滤动作发生在订阅端流量上有些浪费。如果在Topic粒度极多的场景下这种广播式过滤会带来很大的带宽压力。一个常见的调优项是zmq.CONFLATE选项这个选项控制订阅端是否只保留最新消息socket.setsockopt(zmq.CONFLATE, 1)设置CONFLATE1之后订阅端的接收队列只保留最新的消息抛弃积压的旧消息。这个选项对实时行情类的场景特别有效比如价格数据消费者只关心最新价格中间跳过的那些旧价格没有意义。4.2 慢消费者问题PUB-SUB最典型的陷阱慢消费者问题指的是发布端的发送速度远大于订阅端的处理速度导致订阅端队列积压最后内存膨胀甚至OOM。这是PUB-SUB模式里最需要重视的问题因为PUB端根本不关心SUB端处理得怎么样它只管发。我用一个实际测试数据来说明这个问题。在一个测试环境中PUB端以每秒1000条的速度发送消息SUB端模拟慢处理每条消息处理耗时10毫秒也就是说每秒只能处理100条。这样每秒会有900条消息积压在SUB端的接收队列里。默认HWM是1000条队列满后ZMQ还会继续往内核缓冲区发最终表现是好消息是ZMQ帮你在缓冲坏消息是内存消耗持续上升如果积压时间足够长进程内存会飙到数GB。针对这个问题业界通用的策略有几种给SUB端设置合理的RCVHWM宁可丢弃也不要让内存爆掉。然后在应用层做消息序号校验发现序号跳跃就记录日志业务端做好补偿机制。使用CONFLATE功能只保留最新状态丢弃历史积累。前提是业务可以容忍跳变而不是丢失后需要补数。评估处理速度与生产速度的匹配程度。如果处理速度长期跟不上应该加SUB端消费者数量或用PUSH-PULL模式做任务分发而不是用广播模式。这三种策略里前两种是代码层面的调整第三种是架构层面的调整实际项目中应该先做第三种评估。4.3 Topic订阅过滤的正确写法ZMQ的Topic过滤是前缀匹配不是通配符匹配。SUBSCRIBEtopic能匹配到topic1、topic2、topic.abc但匹配不到TOPIC和mytopic。这一点跟消息队列里的通配符订阅规则比如topic.*、topic.#完全不同是新手最容易踩的坑。来看一个典型的多Topic场景假设一个系统里有三类消息要推送socket.setsockopt_string(zmq.SUBSCRIBE, stock.) socket.setsockopt_string(zmq.SUBSCRIBE, weather.)这两行表示订阅所有以stock.和weather.开头的消息。发布端发送消息的时候就要把Topic作为消息前缀拼进消息体socket.send_string(stock.000001.price: 10.5) socket.send_string(weather.city: sunny)这种做法的代价是业务消息需要自定义协议来区分Topic和Payload。ZMQ官方也提供了多帧消息Multi-part Message的方式来解决这个问题把第一帧作为Topic帧第二帧作为Payload帧。实践中我推荐用多帧方式解析起来更清晰也不用在字符串上切割。4.4 订阅者初始连接的消息丢失窗口PUB-SUB模式里有一个非常经典的现象SUB端刚连接上PUB端的头几秒钟可能会丢失一部分消息。原因在于当SUB端执行connect的时候ZMQ底层需要做TCP握手这个握手过程是在后台异步完成的。在握手完成之前SUB端的订阅信息还没到达PUB端PUB端已经发出的消息自然就不会被转发。这个现象不是ZMQ的Bug而是异步连接模型的固有属性。解决办法是如果应用不能接受启动初期的少量丢包需要做一次握手同步。常见做法是让SUB端发送一个握手完成信号给PUB端PUB端确认收到后再开始发送业务数据。在ZMQ里你可以用REQ-REP模式做这个握手通道业务数据走PUB-SUB通道。这种混合模式的架构在实际项目中很常见。5. PUSH-PULL模式的任务流水线实现PUSH-PULL是ZMQ三种模式里最适合做并行任务处理的。它的路径是上游PUSH端把任务发给下游PULL端每个任务只会被一个PULL端收到。如果要实现结果聚合就在PULL端处理后用PUSH socket发给一个最终的PULL端收集器。5.1 任务分发与多Worker并行处理用一个简单的并行计算例子来说明。假设有一个图像处理任务要把一万张图片做灰度转换单个进程处理要花十分钟。用PUSH-PULL模式可以把这一万张图片拆成任务列表一个PUSH端轮流发给五个Worker每个Worker处理两千张总耗时能压缩到两分钟出头。发布任务端代码import zmq context zmq.Context() socket context.socket(zmq.PUSH) socket.bind(tcp://*:5557) for task_id in range(10000): task ftask-{task_id} socket.send(task.encode()) # 发送结束标志 socket.send(bEND)Worker端代码import zmq context zmq.Context() socket context.socket(zmq.PULL) socket.connect(tcp://localhost:5557) while True: task socket.recv().decode() if task END: break # 模拟处理耗时 result fdone-{task}这里有一个非常关键的行为特征PUSH端的消息分发是公平调度负载均衡的。PUSH端在内部维护着一个可用接收端的列表每发一条消息就轮换一个接收端。这样只要下游都能及时接收任务就能比较均匀地分配。这个行为是ZMQ自动完成的不需要你写任何负载均衡逻辑。不过要注意一点这个公平调度是工作窃取式的还是简单轮询式的跟ZMQ版本和底层行为有关。在多数版本里PUSH到多个PULL是轮询分发如果你对负载均衡的均匀性要求极高比如某些任务的处理耗时差异极大建议在业务层实现动态反馈机制比如Worker处理完一条任务后主动通过另一个管道上报我空了PUSH端根据反馈优先给空闲Worker发任务。这就是从任务驱动变成工作者驱动的思路在专业术语里叫工作窃取模式。5.2 PUSH-PULL天然存在的木桶效应PUSH-PULL有一个非常现实的问题整条流水线的吞吐量由最慢的那一段决定。比如PUSH端每秒能发500个任务下游Worker每个每秒能处理100个那配5个Worker刚好是平衡如果其中一个Worker因为机器性能差或者网络抖动每秒只能处理50个整条管线的总吞吐量就会被拉低。为什么是拉低而不是被其他Worker补上因为PUSH端公平轮询它不会根据Worker处理速度动态调整分发比例。如果5个Worker里有一个速度慢一半那分到这个慢Worker上的任务会积压而其他Worker可能已经空闲了。这就会造成整体任务耗时被拉长。解决办法是上面提到的工作者驱动模型Worker处理完一个任务后向PUSH端发起给我下一个任务的请求而不是PUSH端主动往Worker身上塞任务。这个模型下快的Worker自然拿到更多任务慢的Worker自然少拿整条管线的吞吐量就变成由所有Worker的处理能力总和决定而不是被最慢的拖累。在ZMQ里工作者驱动模型可以用DEALER-ROUTER来完美实现这也是下篇要重点展开的内容。这里先说结论如果你在PUSH-PULL模式下遇到负载严重不均不用在PUSH-PULL里死磕直接升级到DEALER-ROUTER模式问题会迎刃而解。5.3 PUSH-PULL做结果汇聚时的连接时间问题任务分发做完Worker处理完最后要把结果汇总到一个收集器。这时候Worker端创建一个PUSH socket连向收集器的PULL socket。在这个阶段有一个值得重点提醒的坑Worker进程刚启动时立即向PUSH socket发送数据前面的几十条甚至几百条消息可能会丢失。原因跟PUB-SUB的订阅者初始连接丢消息问题本质相同底层TCP连接是异步建立的调用send返回成功只能说明数据放进了ZMQ的发送队列不代表已经发到对端网络。在连接还没建立的瞬间发消息这些消息就会在队列里等待如果队列满了新消息就会因为背压被阻塞。规避办法是启动后等连接稳定再发批量的数据。怎么做在Worker启动后发送一个特殊的握手消息比如HELLO确认收集器那边收到了再发正式数据。收集器收到HELLO后回一个ACK。这样做还带来一个额外收益你同时验证了整条链路的连通性。6. 三种模式的对比选型与实战经验三种模式都介绍完了作一个对比总结然后分享一些实操层面的经验和排查技巧。6.1 模式选型速查表通信需求推荐模式核心特征需要注意的点远程调用/一问一答REQ-REP强同步、严格交替REQ和REP都必须遵守交替纪律数据广播/一对多PUB-SUB发布者与订阅者解耦订阅者上线晚会丢消息需要关注慢消费者任务并行分发PUSH-PULL负载均衡流水线最慢节点拖累整体吞吐异步请求路由DEALER-ROUTER下篇支持异步双向适合复杂拓扑上手成本略高双工一对一PAIR下篇双向对等通信只支持两个节点极少用到选型的时候我一般按三个问题走一遍第一通信双方是否需要同步等待结果需要就选REQ-REP。第二是否需要一对多广播、生产者是否要感知消费者数量需要就选PUB-SUB。第三业务模型是否是任务流水线、需要并行处理需要就选PUSH-PULL。还有一点经验三种模式可以组合使用。一个实际项目里服务发现可以用REQ-REP做状态推送用PUB-SUB做批量任务处理用PUSH-PULL做各司其职。ZMQ不限制你只能用一个模式反而鼓励在一个进程里用不同的Socket连接不同的上下游形成一张通信拓扑网络。6.2 模式之外ZMQ的核心特性补充三种模式背后共享的ZMQ特性同样值得强调。ZMQ支持多种传输协议tcp用于网络通信ipc用于本机进程间通信inproc用于同一进程内线程间通信。在线调试或者单测环境里用inproc能免去网络栈的开销测试速度极快。我习惯把核心通信逻辑封装成独立的模块单元测试里全部用inproc协议等集成测试时再切换到tcp这个切换通常只改一行字符串。ZMQ的消息是二进制的、分帧的。它发送和接收的消息是基于帧的不会出现传统TCP那种半包粘包问题。底层帮你把消息切成帧接收端收到的一定是完整的一帧数据。这一点是ZMQ大幅简化开发的核心原因之一。你不需要处理粘包拆包不需要自己定义消息长度字段ZMQ已经在协议层把这些事做完了。ZMQ的所有Socket在应用层面是可以断线重连的。一个REQ socket连一个还没启动的REP服务端REQ端调用connect后立即发消息消息会在队列里等着等REP端启动后才能收到。这个等的过程ZMQ自动完成。这个特性在部署场景里价值巨大因为服务端和客户端的启动顺序不用再严格要求了。6.3 实战中遇到的高频问题排查记录把实践中最常碰到的问题整理成一份速查表每条都是我自己调过的坑。现象可能原因排查方向SUB端收不到任何消息没设置SUBSCRIBE过滤器socket.setsockopt_string(zmq.SUBSCRIBE, )客户端启动后前几条消息丢失TCP异步连接未建立增加握手确认流程REQ端连续send报错违反交替流程检查是否为send-send或recv-recvPUSH端消息全发给了一个Worker其他Worker连接未就绪检查Worker启动顺序增加握手同步进程内存不断上涨接收队列积压HWM超限调低RCVHWM/SNDHWM检查消费速度偶发消息重复重连后消息重发机制业务层做幂等处理增加消息序号网络环境差时收发不稳定默认TCP参数不适合弱网设置TCP心跳参数配合应用层重连这些经验里目前最想重复强调的一点ZMQ的队列和重连机制是用来抗瞬时抖动的不是用来做持久化存储的。不要把ZMQ当消息队列用它的消息在内存里进程一退出就全没了。如果你是想要持久化、消息回溯、分布式事务这类能力请选择专业消息队列ZMQ只负责快、轻、简单。6.4 关于调试与监控的实操建议ZMQ的调试体验在早期版本里不算好因为你看不到内部状态。但用多了之后会发现几个实用工具价值的思路。第一用zmq_socket_monitorAPI监控Socket事件。ZMQ提供了连接事件监控能力可以捕获到CONNECTED、DISCONNECTED、ACCEPTED这类事件。在pyzmq里可以通过zmq.SocketMonitor来获取。这个监控能力对排查为什么消息没发出去非常重要它能告诉你连接的建立和断开时刻把网络层的问题和应用层的问题区分开。第二合理设置ZMQ_LINGER。它控制Socket关闭时发送队列里未发完的消息要等多久才丢弃。默认情况下LINGER值是-1表示永久等待这会让进程在关闭时卡住设置为0是立即丢弃。生产环境建议设置为100到500毫秒给没发完的消息一个短暂的善后时间又不至于拖慢进程退出。第三端到端日志链。ZMQ没有业务层面的追踪机制所以在消息体里带上消息ID和时间戳非常重要。遇到消息丢失或乱序时光靠ZMQ本身看不出问题需要把收发两端的消息ID日志拉出来做比对找到断裂点。这是排查ZMQ生产问题最有效的手段没有之一。7. 最后分享一点关于ZMQ选型的个人体会做了几年通信中间件相关的项目之后我对ZMQ的态度经历了三个阶段的变化。最开始觉得它不过是个封装好的Socket库没什么了不起后来在项目里用REQ-REP和PUB-SUB解决了几个传统Socket很难缠的痛比如断线重连、粘包处理、一对多广播开始觉得它确实有价值再到现在当面对新的通信需求时我不会再问这个能用ZMQ做吗而是会先想ZMQ的哪种模式最贴合这个业务语义。我的实际体会是ZMQ的学习曲线在通信领域里算是比较平滑的它真正的难点不在API而在模式思维。你需要适应Socket自带通信语义这种思维转换从传统Socket的端点思维切换到模式思维。一旦适应了这种转换你会发现用它搭一套复杂的通信拓扑可能比用传统Socket写一个简单的TCP客户端还要快。下篇文章里我会把DEALER-ROUTER、PAIR模式和ZMQ的多部分消息、代理模式、消息加密这些进阶内容展开讲。到时候会重点讲的场景是如何用DEALER-ROUTER实现异步RPC、动态Worker管理以及如何用ZMQ的代理机制组建一个星型或树型的通信网络。到时候继续聊。