最近在做一个分布式采集的中间件改造几个服务之间需要频繁传递任务消息最开始图省事直接上了HTTP接口轮询结果超时重试、连接风暴、数据丢失轮着来。后来换成了ZMQZeroMQ这套消息通信库才算把这块理顺。这个系列打算把我对ZMQ通信模式的理解完整梳理一遍本篇是上篇重点讲最常用的三种基础模式请求-应答REQ/REP、发布-订阅PUB/SUB、推拉管道PUSH/PULL以及它们各自适合什么场景、有哪些坑配合可运行的代码示例方便你直接照着复现。如果你正在做微服务通信、分布式任务调度、数据采集管道这类东西又不想引入Kafka或RabbitMQ这种重量级中间件那ZMQ会是一个非常合适的选择。它足够轻没有独立Broker进程功能又足够硬核理解好通信模式就等于拿到了它的使用钥匙。1. 从裸socket到消息总线的跳跃ZMQ到底解决了什么问题1.1 传统socket开发中最让人头疼的三件事很多人刚接触ZMQ时容易把它当成一个普通的网络库但实际上它解决的是如何优雅地建立分布式消息通道这个层面的问题。用裸socket写通信时你大概率会被三件事折磨。第一是连接管理。两个进程要通信你得手动处理建立连接、断开重连、心跳保活。连接断了他不会自己恢复你需要写一堆重试逻辑。第二是消息边界。TCP是字节流你发了hello和world对端收到的可能是helloworld、也可能是hel和loworld你得自己设计报文头、长度字段、粘包拆包逻辑。第三是架构灵活性。裸socket天然是1对1的想要1对多广播、多对多消息路由那得自己实现一套注册发现机制。ZMQ把这三件事全部收纳了。它内置了断线重连机制你写的代码里根本不需要关心连接状态——它自己会悄悄重连。它面向消息而非字节流每条消息有天然边界不存在粘包拆包问题。它提供多种通信模式就像给消息传输定义了不同规矩你照着规矩连就行。一句话理解socket给你的是字节管道ZMQ给你的是消息总线后者自带连接管理、消息分帧和多种路由策略。1.2 ZMQ的核心抽象消息、端点与上下文使用ZMQ需要先理解三个基础对象Context、Socket和Endpoint。Context是ZMQ的总环境所有socket都必须在一个Context中创建类似进程级的资源管理器。在pyzmq中通常一行代码搞定ctx zmq.Context()。每个socket通过socket_type指定模式比如zmq.REQ、zmq.PUB。Endpoint是socket绑定的地址ZMQ支持多种传输协议最常用的是tcp://跨主机、ipc://同一主机进程间、inproc://同一进程内线程间。这里要特别强调一个设计理念ZMQ的socket不像传统socket那样区分服务端和客户端。一个socket可以bind绑定也可以connect连接而且规则很灵活——不是只有服务端才能bind客户端也能bind服务端也能connect。理解这一点对后面的模式会很关键。1.3 环境准备安装与第一个连接动手之前先把环境装上。Python环境用pip安装pyzmqpip install pyzmq安装完成后写一个最简单的请求-应答通信验证环境是否正常。这是服务端import zmq context zmq.Context() socket context.socket(zmq.REP) socket.bind(tcp://*:5555) while True: message socket.recv() print(f收到请求: {message.decode()}) socket.send(bworld)这是客户端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()})先运行服务端再运行客户端客户端会打印出收到回复: world。这批代码有一个很值得品味的点客户端connect后并没有做任何握手逻辑服务端也没有accept。ZMQ把连接过程完全封装了你只管收发消息。我第一次跑通时确实有种这也太省事了吧的感叹。2. 请求-应答模式REQ/REP的严格一问一答2.1 REQ/REP的工作机制与锁死现象请求-应答模式是最容易上手的模式它模拟的是函数调用的语义客户端发一条请求服务端处理完后必须回一条响应一问一答节奏非常严格。REQ和REP的严格性体现在它们各自的状态机上。REQ socket在send之后不能再次send必须等recv拿到响应后才能进入下一轮发送。REP socket在recv到请求后不能再次recv必须send返回响应后才能接收下一条请求。这个设计保证了消息不会交错但也带来一个新手最常踩的坑——锁死。我见过很多人在REQ端连发两条消息代码大概是这样的socket.send(brequest-1) # 忘记 recv socket.send(brequest-2)程序会直接抛异常或者卡死。REQ在第一次send后状态变为等待响应此时再send就违反了状态机。另一种锁死情况是REP端处理完请求后忘了send响应那REQ端会一直阻塞在recv。所以使用REQ/REP时一定要记住REQ的send必然对应REP的recvREP的send必然对应REQ的recv这是一条严格闭环。2.2 代码实测同步请求的服务端与客户端下面是更丰富一点的示例模拟一个加法计算服务。服务端接收两个整数返回它们的和。服务端代码import zmq context zmq.Context() socket context.socket(zmq.REP) socket.bind(tcp://*:5556) print(加法服务已启动监听端口 5556) while True: # 接收请求格式为 bnum1 num2 request socket.recv() text request.decode() parts text.split() if len(parts) ! 2: socket.send(binvalid request) continue try: a, b int(parts[0]), int(parts[1]) result a b socket.send(str(result).encode()) except ValueError: socket.send(binvalid number)客户端代码import zmq context zmq.Context() socket context.socket(zmq.REQ) socket.connect(tcp://localhost:5556) for i in range(10): socket.send(f{i} {i * 2}.encode()) reply socket.recv() print(f请求: {i} {i * 2}响应: {reply.decode()})运行后客户端会顺序打印10次计算结果。注意这里每次send后必须等recv完成服务端处理慢时客户端就会等待这正是同步调用的一种体现。2.3 REQ/REP逃不出的限制与扩展思路REQ/REP简单可靠但限制也很明显同步、阻塞、一对一。如果服务端处理一条请求需要2秒那这个服务端同一时间就只能服务一个客户端其他客户端的请求全部排队。面对并发场景很多人第一反应是给客户端开多线程每个线程独立REQ socket。这个做法有效但线程数量一多连接数也会暴增并不优雅。更合适的做法是引入ZMQ的ROUTER和DEALER socketROUTER接收客户端请求发给一群DEALER worker处理worker把结果传回ROUTER再转给对应客户端。这就是经典的经纪人模式。不过ROUTER/DEALER属于进阶内容本篇先点到为止下篇会专门展开。先记住结论如果你的REQ/REP遇到了性能瓶颈别在REP端堆线程要考虑升级成ROUTER/DEALER架构。3. 发布-订阅模式PUB/SUB的广播世界3.1 PUB/SUB的消息流向与订阅过滤机制发布-订阅模式解决的是一对多广播问题和REQ/REP完全不同发布者PUB只负责发不知道也不关心消息有没有人收到订阅者SUB只负责收不需要给发布者任何应答。这里有一个特别反直觉的设计SUB端必须设置订阅过滤器才能收到消息。如果你只创建一个SUB并connect不设置任何过滤条件就recv程序会一直卡着收不到任何数据。原因在于ZMQ的设计者把是否订阅所有消息这个选择权交给了SUB端。设置过滤器的方法是在SUB socket上调用setsockopt(zmq.SUBSCRIBE, b)空字节串表示订阅所有消息。也可以传前缀比如bweather表示只接收以weather开头的消息。这个消息过滤是在哪里发生的在PUB端。SUB端即使设了过滤器也不会把过滤条件传给所有下游而是在建立连接时把自己的订阅信息告诉PUB。PUB会根据所有订阅者的过滤器决定是否把消息塞进对应的连接队列。这样做的好处是节省了网络带宽——不相关的消息根本不会发出。3.2 代码实测天气播报的发布订阅写一个天气播报的例子。发布者持续发布包含城市和温度的天气预报消息订阅者分为两类一类订阅所有城市的消息另一类只订阅Shanghai的消息。先看发布端import zmq import time import random context zmq.Context() socket context.socket(zmq.PUB) socket.bind(tcp://*:5557) print(天气发布服务已启动) cities [Beijing, Shanghai, Guangzhou, Shenzhen] while True: city random.choice(cities) temp random.randint(-5, 35) message f{city} {temp}度.encode() socket.send(message) print(f发布: {message.decode()}) time.sleep(1)订阅端订阅全部信息import zmq context zmq.Context() socket context.socket(zmq.SUB) socket.connect(tcp://localhost:5557) socket.setsockopt(zmq.SUBSCRIBE, b) # 订阅所有消息 while True: message socket.recv() print(f收到: {message.decode()})另一个订阅端只订阅Shanghai的消息import zmq context zmq.Context() socket context.socket(zmq.SUB) socket.connect(tcp://localhost:5557) socket.setsockopt(zmq.SUBSCRIBE, bShanghai) # 只订阅Shanghai前缀 while True: message socket.recv() print(f上海天气: {message.decode()})运行发布端再运行任意订阅端你会看到订阅端稳定收到消息。注意顺序必须先启动发布端再启动订阅端否则订阅端刚启动时可能错过前几条消息。这是因为SUB连接PUB需要时间在连接建立完成前PUB发出的消息会被丢弃。3.3 慢消费者与消息丢弃PUB/SUB最常踩的坑PUB/SUB最经典的坑有两个。第一个是慢消费者问题。假设PUB每秒发1000条消息某个SUB处理能力只有每秒100条它处理不过来。ZMQ的处理策略不是无限等待而是设置了高水位HWMHigh Water Mark。当某个SUB的接收队列堆积超过HWM值默认1000条时PUB会直接丢弃该订阅者的溢出消息。这个特性在很多场景下是好事——它避免了某个慢消费者拖垮整个发布端。但如果你恰好需要消息必须全部送达PUB/SUB就不能满足要求你需要更可靠的消息系统或者自己设计ACK机制。我在实际项目里处理这个问题的做法是给重要消息加序号订阅端检测到序号跳变时主动向发布端请求补偿这个模式其实就是可靠发布订阅的简化版工作得很好。第二个坑是订阅者迟到。在ZMQ里如果SUB先启动、PUB后启动SUB将收不到PUB启动之前的所有消息。它不像Kafka那样有持久化日志PUB发出的消息如果当时没有订阅者连接消息就直接消失了。所以如果你需要一个消息必须被消费的业务场景PUB/SUB从一开始就不该是首选。4. 推拉模式PUSH/PULL的数据管道4.1 任务分发的公平调度原理推拉模式PUSH/PULL最初是为处理管道设计的上游PUSH把任务推给下游PULLPULL处理完后再把结果PUSH给下一个节点。它和REQ/REP的区别在于PUSH端根本不在乎有没有人接收处理只管推送和PUB/SUB的区别在于每条消息只给一个PULL接收者而不是广播给所有人。当多个PULL节点连接同一个PUSH时PUSH采用的是公平排队round-robin策略它维护一个接收者列表依次把消息轮流发送给每个接收者。如果其中一个PULL节点断开它会自动从列表移除之后消息就只分给仍然在线的节点。这个公平分发特性非常适合做并行任务分发。我在数据采集场景里用过这种结构一个任务源PUSH把成千上万个URL推给多个采集worker所在的PULL节点每个worker拿到的URL数量大致相同采集完成后各自处理。4.2 代码实测三阶段处理管道用一个三阶段管道来演示。第一个进程是任务源生成100个任务并PUSH出去第二个进程是处理节点PULL任务、把字符串转大写、再PUSH到下一个阶段第三个进程是结果收尾PULL并打印结果。先看任务源import zmq context zmq.Context() socket context.socket(zmq.PUSH) socket.bind(tcp://*:5558) print(任务源启动开始推送任务) for i in range(100): task ftask-{i}.encode() socket.send(task) if i % 20 0: print(f已推送: {task.decode()})处理节点这里我建议你可以启动两个实例来验证负载均衡import zmq import time context zmq.Context() receiver context.socket(zmq.PULL) receiver.connect(tcp://localhost:5558) sender context.socket(zmq.PUSH) sender.bind(tcp://*:5559) print(处理节点启动等待任务...) while True: task receiver.recv() processed task.decode().upper().encode() sender.send(processed) print(f处理完成: {processed.decode()}) time.sleep(0.01) # 模拟处理耗时任务源和处理节点可以放在同一台机器上但ZMQ在这种跨端口的连接场景下到底谁bind谁connect其实完全看设计。上面示例中任务源bind了5558处理节点connect过去再bind自己的输出端口5559。你可以开两个处理节点进程它们会分别从5558拿任务两个进程之间呈现轮流拿任务的现象这正好验证了round-robin分发。结果收尾进程import zmq context zmq.Context() socket context.socket(zmq.PULL) socket.connect(tcp://localhost:5559) print(结果收集器启动) count 0 while count 100: message socket.recv() print(f最终结果: {message.decode()}) count 1先启动结果收尾再启动处理节点最后启动任务源就能看到一个完整的管道跑通。如果把任务源的消息量增大你还会发现某个处理节点偶尔会比另一个收到的任务更多这取决于ZMQ内部对每个连接的排队情况但总体趋势是均衡的。4.3 管道模式的边界背压与可靠性PUSH/PULL适合做数据管道但它同样有边界条件。第一个是背压问题如果下游处理速度远慢于上游推送速度PUSH端的发送队列会持续膨胀。ZMQ在达到HWM后PUSH端的send会进入阻塞状态这算是ZMQ少有的天然限流机制。阻塞带来的连锁反应是上游任务源也会变慢但至少不会丢消息。相比之下PUB/SUB是直接丢弃PUSH/PULL是阻塞等待这是两种截然不同的行为差异。第二个是可靠性问题。PUSH/PULL不保证消息传输可靠性。如果某个PULL节点在收到消息后、处理完成前崩溃了那条消息就丢失了。解决思路通常是在业务数据里增加任务编号配合一个存储层做幂等控制或者使用可靠消息中间件来兜底。ZMQ本身是智能的传输层不是一个自带分布式事务的消息平台。5. 三种模式怎么选从业务场景倒推通信方案5.1 一张对照表看清三者的差异了解单个模式之后再来看整体我梳理了一张对照表直接从业务场景倒推通信方案参考价值很高。维度REQ/REPPUB/SUBPUSH/PULL消息流向双向严格一问一答单向一对多广播单向一对一分发发送方状态必须等待应答不关心谁接收不关心接收结果接收方数量通常一对一多个订阅者多个接收者消息是否可能丢失不会阻塞等应答会慢消费者和迟到订阅者会丢可能节点崩溃则消息消失典型场景远程函数调用、命令下发状态广播、日志分发、行情推送任务分发、流水线处理负载均衡能力弱单服务端串行处理无所有订阅者都收同一份强按队列均衡分配这张表里最值得关注的是消息是否可能丢失这一行。在实际生产环境里这是选型的第一决定因素。我自己的经验是能容忍丢失且希望多接收方的用PUB/SUB需要严格同步响应的用REQ/REP需要任务分流且可容忍极少丢失的用PUSH/PULL。5.2 混合使用一个服务同时扮演多个角色前面讲的是三个独立模式但真实世界里服务往往不是只用一种socket的。ZMQ允许你在同一个进程里创建多个socket分别扮演不同角色这简直是为复杂的分布式系统量身定制的。举个例子一个采集服务可以同时做三件事用REP master接口接收管理者下发的采集任务用PUSH把URL推给下游解析worker再用SUB订阅一个全局状态频道接收暂停/恢复指令。三个socket互不干扰协调工作。这种组合能力是ZMQ相比很多消息中间件更灵活的地方也是我后来越来越喜欢它的原因。混合使用时要注意一件事socket类型和模式必须严格匹配不能拿一个REP去连另一个REP也不能让PUSH端去recv。ZMQ会在消息层面主动拒绝不兼容的模式配对报错信息也足够明确遇到报错时先检查一下两端socket类型是否匹配大部分问题都能快速定位。5.3 上篇小结与后续方向预告写到这里ZMQ的三种基础通信模式应该已经很清晰了。REQ/REP是一问一答的同步调用PUB/SUB是广播式的消息分发PUSH/PULL是均衡式的数据管道。它们覆盖了分布式系统中绝大多数的通信需求而且代码量极低对比手写socket或者部署一个完整消息中间件性价比非常高。我个人在实际使用中最大的体会是用ZMQ一定不要惯性思维。很多从HTTP或者传统消息队列转过来的开发者容易下意识地去找服务端客户端请求地址这些概念但ZMQ的模式思维是角色决定行为——同一台机器、同一个端口可以在一套架构里既是服务端又是客户端。想通这一点ZMQ的世界就会豁然开朗。还有一点值得提在跑示例时如果发现消息延迟或丢失先检查是不是忘了设置SUBSCRIBE过滤器再检查HWM和LINGER设置。HWM决定队列上限LINGER影响socket关闭时是否等待未发送消息这两个都是ZMQ调优里的高频参数。ZMQ的进阶武器还没讲完面向多对多路由的ROUTER/DEALER、进程内通信的inproc优化、发布订阅的可靠代理模式、多消息帧的处理技巧这些内容都留到下篇再和各位细聊。每一种模式背后都有非常实用的生产场景把基础打牢之后进阶内容会顺畅很多。