用 ZooKeeper 实现分布式队列

队列在服务端的样子

实测 · 本机执行输出

  /q 子节点: ['entry-010-0000000005', 'entry-100-0000000000', 'entry-100-0000000001', 'entry-100-0000000002', 'entry-100-0000000003', 'entry-100-0000000004']
  取出顺序(priority 小的先出,同优先级先进先出): ['urgent', 'job-0', 'job-1', 'job-2', 'job-3', 'job-4']
  队列空时 get() 返回: None

和顺序节点那篇的结论一致:序号由服务端分配,所以多个生产者同时入队也不会冲突。

普通 Queue:取出即删除

实测 · 本机执行输出

  队列最终为空;200 条中被取出 200,处理完成 199,被取出但没有完成(被杀的消费者手里那条)= ['m020']

出队是「读最小节点 + 删除」一步完成,消息一旦出队就只存在于消费者内存里。消费者在处理中崩溃,没有任何地方还留着它。能接受丢失(比如统计、缓存刷新通知)才用这种。

LockingQueue:至少一次

实测 · 本机执行输出

  30 条:被取出的消息数 30,被确认 30;被取出了 2 次的:['j12'];取出却没确认的:[]
    j12 ['-0.1s C0 GOT', '+3.8s C1 GOT', '+4.1s C1 ACK']

消息先被「锁」住(一个临时节点,属于消费者的会话),处理完调用 consume() 才真正删除。消费者崩溃后会话过期,锁节点消失,消息重新可见。结果是不丢但可能重复:如果消费者处理完了、还没来得及 consume 就崩溃,这条会被处理第二次。

吞吐

实测 · 本机执行输出

  Queue:入队 305 条/s,出队 308 条/s;LockingQueue:入队 295 条/s,出队+确认 143 条/s
  /bq 里的单条消息大小限制:ZooKeeper 默认每个 znode ≤ 1 MiB(见第 1 篇);队列里所有未消费的消息都在内存与快照里(第 10 篇)

队列里未消费的每条消息都是内存里的 znode、也在快照里(见DataTree),积压成千上万条会拖慢 get_children。堆积量大、吞吐高的场景请用 Kafka / RabbitMQ 之类,ZooKeeper 队列适合少量、低频、需要强一致排序的协调任务。

分步演示

分布式队列

本步的对象

本文用到的工具与边界

本机 macOS 27 / ZooKeeper 3.9.6(brew 安装的 jar,用 java -cp 直接启动自己的实例,OpenJDK 17 Temurin,JVM 堆 128 MiB)/ 本机回环端口 7300+,数据在临时目录;默认 tickTime=2000;计时用 time.perf_counter,数字每次运行会有小幅抖动。kazoo 2.11.0 的 Queue 与 LockingQueue;消费者是独立子进程;会话超时 3 s;吞吐是单进程单连接的同步调用,不是集群上限
没有验证:PriorityQueue 之外的 kazoo 队列(LockingQueue 的优先级)、几万条积压时 get_children 的耗时、多消费者并发取同一条的竞争细节、与 Kafka / RabbitMQ 的对比

下一篇 · ZOOKEEPER 83

用 ZooKeeper 实现分布式屏障