在 Python 中操作 ZooKeeper（通常简称为 ZK）主要依赖于第三方库，ZooKeeper 是一个分布式协调服务，常用于配置管理、命名服务、分布式同步等场景。
以下是关于在 Python 中使用 ZooKeeper 的完整指南：
常用 Python 库
目前最主流、维护较好的 Python ZooKeeper 客户端库是：
✅ kazoo
- GitHub: https://github.com/python-zk/kazoo
- 特点:
- 纯 Python 实现，易于安装和使用。
- 支持异步（asyncio）和同步两种模式。
- 社区活跃,文档较完善。
- 推荐用于新项目。
其他库（较少使用）
pyzk: 基于 C 扩展，性能稍好但安装复杂。zookeeper: 较老，维护较少。
推荐使用
kazoo，下面以kazoo为例进行说明。
安装
pip install kazoo
基本使用示例
1 同步客户端
from kazoo.client import KazooClient
from kazoo.retry import KazooRetry
# 创建 ZooKeeper 客户端
zk = KazooClient(hosts='127.0.0.1:2181')
zk.start()
try:
# 创建节点（如果不存在）
zk.ensure_path('/my/test/path')
# 写入数据
zk.set(&
#39;/my/test/path', b'hello world')
# 读取数据
data, stat = zk.get('/my/test/path')
print(f"读取到的数据: {data.decode()}")
# 创建临时节点（会话结束后自动删除）
zk.create('/my/temp/node', b'temp data', ephemeral=True)
# 监听节点变化
@zk.ChildrenWatch('/my/test/path')
def on_children_changed(children):
print(f"子节点变化: {children}")
# 保持运行以接收事件
import time
time.sleep(10)
finally:
zk.stop()
zk.close()
2 异步客户端（基于 asyncio）
import asyncio
from kazoo.client import KazooClient
async def main():
zk = KazooClient(hosts='127.0.0.1:2181')
await zk.start()
try:
await zk.ensure_path('/async/test')
await zk.set('/async/test', b'async data')
data, stat = await zk.get('/async/test')
print(f"异步读取: {data.decode()}")
finally:
await zk.stop()
await zk.close()
asyncio.run(main())
常用操作
| 操作 | 方法 | 说明 |
|---|---|---|
| 创建节点 | zk.create(path, value=b'', acl=None, ephemeral=False, sequence=False) |
创建持久/临时/顺序节点 |
| 读取数据 | zk.get(path) |
返回 |
| 写入数据 | zk.set(path, value) | 更新节点数据 |
| 删除节点 | zk.delete(path, version=-1) | 删除节点 |
| 获取子节点 | zk.get_children(path) | 返回子节点列表 |
| 检查节点是否存在 | zk.exists(path) | 返回 stat 或 None |
| 监听子节点变化 | @zk.ChildrenWatch(path) | 装饰器方式监听 |
| 监听数据变化 | @zk.DataWatch(path) | 装饰器方式监听 |
节点类型
- 持久节点（Persistent）：节点创建后一直存在，除非手动删除。
- 临时节点（Ephemeral）：与客户端会话绑定，会话断开后自动删除，常用于服务注册与发现。
- 顺序节点（Sequential）：节点名后自动追加单调递增的数字，如
/my/node0000000001。
典型应用场景
1 服务注册与发现
import uuidimport timefrom kazoo.client import KazooClientzk = KazooClient(hosts='127.0.0.1:2181')zk.start()# 注册服务service_id = str(uuid.uuid4())zk.create(f'/services/myapp/{service_id}', b'192.168.1.100:8080', ephemeral=True)# 发现服务services = zk.get_children('/services/myapp')print(f"可用服务: {services}")zk.stop()zk.close()
2 分布式锁
from kazoo.client import KazooClient
from kazoo.recipe.lock import Lock
zk = KazooClient(hosts='127.0.0.1:2181')
zk.start()
lock = Lock(zk, '/my-lock')
with lock:
print("获得锁，执行临界区代码...")
# 临界区代码
print("释放锁")
zk.stop()
zk.close()
注意事项
- 会话超时：ZooKeeper 客户端与会话断开后，临时节点会被删除，合理设置
session_timeout_ms。 - 重试机制：网络抖动时，建议使用
KazooRetry配置重试策略。 - ACL 权限：生产环境建议配置 ACL 控制访问权限。
- 连接状态：监听
zk.add_listener事件以处理连接断开/重连。
def my_listener(event):
if event.type == 'CONNECTED_STATE':
print("已连接")
elif event.type == 'DISCONNECTED_STATE':
print("已断开")
zk.add_listener(my_listener)
参考资料
- Kazoo 官方文档：https://kazoo.readthedocs.io/
- ZooKeeper 官方文档：https://zookeeper.apache.org/doc/current/
首发原创文章，作者：世雄 - 原生数据库架构专家，如若转载，请注明出处：https://idctop.com/article/477280.html
