11.7.4 集群协调与分布式队列实践

4. 集群协调与分布式队列实践#

场景导语#

本节以 ZooKeeper 实现“集群协调 + 简易分布式队列”为例，覆盖原理草图、节点设计、命令操作、排错与练习，便于快速落地验证。

原理草图（集群协调 + 队列）#

文章图片

环境准备与安装示例#

以单机 ZooKeeper 为演示基础，集群部署可参考本章“安装部署与集群搭建”。

# 1) 安装（基于二进制包）
sudo useradd -r -s /sbin/nologin zookeeper
wget https://archive.apache.org/dist/zookeeper/zookeeper-3.8.1/apache-zookeeper-3.8.1-bin.tar.gz
tar -xzf apache-zookeeper-3.8.1-bin.tar.gz -C /opt/
ln -s /opt/apache-zookeeper-3.8.1-bin /opt/zookeeper
mkdir -p /data/zk/{data,log}

# 2) 配置
cat >/opt/zookeeper/conf/zoo.cfg <<'EOF'
tickTime=2000
initLimit=10
syncLimit=5
dataDir=/data/zk/data
dataLogDir=/data/zk/log
clientPort=2181
autopurge.snapRetainCount=5
autopurge.purgeInterval=1
EOF

# 3) 启动
/opt/zookeeper/bin/zkServer.sh start
/opt/zookeeper/bin/zkServer.sh status

集群协调典型模式#

  • 成员管理：通过临时节点维护在线实例列表，结合 watch 感知上下线变化。
  • 主从切换：使用临时顺序节点竞争 leader，保障自动故障转移。
  • 分片协调：用持久节点存储分片元数据，动态调整任务分配。
  • 状态同步：持久节点保存配置版本，配合 CAS 更新。

分布式队列实现方式#

  • 简单队列：生产者创建持久顺序节点，消费者按序获取并删除节点。
  • 消费者分组：以子目录区分消费组，避免重复消费。
  • 并行消费：消费者对节点加锁或抢占处理，配合超时与重试。
  • 幂等保证：业务侧引入去重 key 与处理日志，避免重复执行。

实践步骤与关键节点设计#

  • /queue/items：生产者写入持久顺序节点 item-0000000001。
  • /queue/consumers：记录消费者信息与处理状态。
  • /queue/locks：用于抢占处理权的临时节点。
  • /queue/meta：队列参数（并发数、最大长度、清理策略）。

核心命令与流程演示（可执行）#

# 1) 连接 ZooKeeper
/opt/zookeeper/bin/zkCli.sh -server 127.0.0.1:2181

# 2) 初始化队列节点
create /queue ""
create /queue/items ""
create /queue/consumers ""
create /queue/locks ""
create /queue/meta "maxLen=1000,cleanInterval=60s"

# 3) 生产者写入任务（持久顺序节点）
create -s /queue/items/item- "task=backup;id=1001;ts=1700000000"
create -s /queue/items/item- "task=backup;id=1002;ts=1700000001"

# 4) 消费者查看队列（按序）
ls /queue/items

# 5) 消费者抢占锁（临时节点）
# 假设要处理 item-0000000001
create -e /queue/locks/item-0000000001 "consumer=worker-1"

# 6) 成功后处理并删除任务
get /queue/items/item-0000000001
delete /queue/items/item-0000000001
delete /queue/locks/item-0000000001

# 7) 记录消费者状态
create /queue/consumers/worker-1 "last=item-0000000001;status=ok"

命令解释要点
- create -s：创建“持久顺序节点”，用于队列有序性。
- create -e：创建“临时节点”，用于锁与成员检测。
- ls：查看子节点列表，可用于获取最小序号任务。
- get/delete：处理并删除任务，确保消费完成。

并行消费示例（简单锁抢占策略）#

# consumer-1
create -e /queue/locks/item-0000000002 "consumer=worker-1"

# consumer-2
create -e /queue/locks/item-0000000002 "consumer=worker-2"
# 预期：只有一个成功，另一个会报错 NodeExists

运维要点与风险控制#

  • 节点数量控制：避免积压导致节点数膨胀，设置最大长度与清理策略。
  • 性能优化：批量读取与删除，减少频繁 watch。
  • 容灾恢复：消费处理前写入处理日志，失败后可重放。
  • 监控指标：延迟、队列长度、活跃连接、会话超时、watch 数量。

故障排查示例#

# 1) 连接失败
# 现象：ConnectionLoss / SessionExpired
# 排查：检查服务状态与端口监听
/opt/zookeeper/bin/zkServer.sh status
ss -lntp | grep 2181

# 2) 锁创建失败（NodeExists）
# 现象：抢占锁失败
# 处理：监听前序节点或重试
ls /queue/locks
# 清理异常锁（确认消费节点已下线）
delete /queue/locks/item-0000000002

# 3) 队列积压
# 现象：items 节点数量持续增长
# 排查：消费端是否存活、锁是否泄露
ls /queue/items
ls /queue/locks

适用场景与边界#

  • 适合 轻量级协调与小规模任务队列，实时性高、数据量小。
  • 不适合 高吞吐大规模消息 场景，应迁移 Kafka/RocketMQ。

练习#

  1. 编写脚本批量创建 50 个 item- 任务，并验证消费顺序正确。
  2. 模拟消费者宕机，观察临时锁节点是否自动删除。
  3. 将 maxLen 设置为 10，编写清理脚本仅保留最新 10 个任务。