基于python操縱zookeeper介紹
by:授客 QQ:1033553122
測試環境
Win7 64位
Python 3.3.4
kazoo-2.6.1-py2.py3-none-any.whl（windows）
kazoo-2.6.1.tar.gz (linux)
https://pypi.org/project/kazoo/#files
zookeeper-3.4.13.tar.gz
下載下傳位址：
http://zookeeper.apache.org/releases.html#download
https://www.apache.org/dyn/closer.cgi/zookeeper/
https://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper/
代碼實踐
kazooStudy.py
#!/usr/bin/env python 3.4.0
#-*- encoding:utf-8 -*-
__author__ = 'shouke'
import threading
import time
from kazoo.client import KazooClient
from kazoo.client import KazooState
from kazoo.retry import KazooRetry
def restart_zk_client():
'''重新開機zookeeper會話'''
global zk_client
global zk_conn_stat
try:
zk_client.restart()
except Exception as e:
print('重新開機zookeeper用戶端異常：%s' % e)
zk_conn_stat = 0 # zookeeper連接配接狀态 1-LOST 2-SUSPENDED 3-CONNECTED/RECONNECTED
def zk_conn_listener(state):
'''zookeeper連接配接狀态監聽器'''
global zk_conn_stat
if state == KazooState.LOST:
print('zookeeper connection lost')
zk_conn_stat = 1
# Register somewhere that the session was lost
thread = threading.Thread(target=restart_zk_client)
thread.start()
elif state == KazooState.SUSPENDED:
print('zookeeper connection dicconnected')
zk_conn_stat = 2
# Handle being disconnected from Zookeeper
else:
zk_conn_stat = 3
print('zookeeper connection cconnected/reconnected')
# Handle being connected/reconnected to Zookeeper
# 螢幕
# 當節點有變化、節點被删除時，将以多線程的方式調用以參數形式傳遞給get()、exists()的監視函數，監視函數将會接收到一個WatchedEvent執行個體
def event_listener(event):
print(event)
if __name__ == '__main__':
# 建立連接配接
zk_client = KazooClient(hosts='127.0.0.1:2181')
zk_client.add_listener(zk_conn_listener) # 添加監聽器，監聽連接配接狀态
zk_client.start() # 初始化到zk的連接配接，可以設定逾時時間 zk_client.start(timeout=15) 預設15秒
print('zk_client state：', zk_client.state) # 檢視連結狀态
# 建立節點
# ensure_path() 遞歸建立path中不存在的節點，但是不能為節點設定資料，僅ACL.
zk_client.ensure_path('/node1')
# 建立永久節點
# create建立節點的同時，可為節點設定資料，要求path路徑必須存在
if not zk_client.exists('/node1/subNode1'):
zk_client.create('/node1/subNode1', b'sub node1')
# 建立臨時節點
# 注意：會話丢失、重新開機會話會導緻zookeeper删除重新開機會話前建立的臨時節點
if not zk_client.exists('/node1/subNode2'):
zk_client.create('/node1/subNode2', b'sub node2', ephemeral=True)
# 建立有序臨時節點
zk_client.create('/node1/subNode', b'sub nodexxxx', ephemeral=True, sequence=True)
# 讀取資料
# 判斷節點是否存在
if zk_client.exists('/node1'): # 如果傳回值為None則表示不存在給定節點
print('存在節點node1,節點路徑/node1')
# 擷取節點相關資料
data, stat = zk_client.get('/node1')
if stat:
print("Version: %s, data: %s" % (stat.version, data.decode("utf-8")))
# 擷取給定節點的子節點
children = zk_client.get_children('/node1')
print('node1子節點 有 %s 子節點，節點名稱為: %s' % (len(children), children))
print('/ 子節點', zk_client.get_children('/'))
# 更新節點
# 更新節點資料
zk_client.set("/node1/subNode2", b"some new data")
# 删除節點 recursive參數可選，遞歸删除節點資料
zk_client.delete("/node1", recursive=True)
# 重試指令
try:
result = zk_client.retry(zk_client.get, "/node1/subNode3")
print(result)
# 自定義重試
# max_tries 出錯最大重試次數， ignore_expire False-重試的時候忽略會話過期，否則不忽略
kr = KazooRetry(max_tries=3, ignore_expire=False)
result = kr(zk_client.get, "/node1/subNode3")
except Exception as e:
print('/node1/subNode3 不存在，是以會運作出錯')
# 釋放用戶端占用資源，移除連接配接
zk_client.stop()
# zk_client.stop() 會導緻zk_client連接配接狀态變成 LOST，進而觸發線程調用函數 restart_zk_client，
# 該函數未執行完成的情況下，如果馬上執行類似get,create等函數，會導緻運作出錯
#
while zk_conn_stat != 3:
continue
else:
i = 0
while i < 3000:
if i % 200 == 0:
time.sleep(2)
print('建立新節點')
zk_client.ensure_path('/node1')
zk_client.ensure_path('/node1/subNode2')
zk_client.create('/node1/subNode', b'sub nodexxxx', ephemeral=True, sequence=True)
zk_client.set('/node1/subNode2', b'new data')
i += 1
# 關閉用戶端前必須先調用stop,否則會報錯
# 關閉用戶端
zk_client.close()
print('運作出錯：%s' % e)
monitor.py
#!/usr/bin/env python
zk = KazooClient(hosts='10.118.52.26:2181')
zk.start()
@zk.add_listener
def my_listener(state):
print('LOST')
print('SUSPENDED')
pass
print('CONNECTED')
children = zk.get_children('/node1',watch=event_listener)
print('node1 has %s children with names %s' % (len(children), children))
# 更進階監視api
# 監視子節點的編号
@zk.ChildrenWatch('/node1')
def watch_children(children):
print("Children are now: %s" % children)
# 監視節點資料變更
@zk.DataWatch("/node1/subNode2") #
def watch_node(data, state):
"""監視節點資料是否變化"""
if state:
print("Version:", state.version, "data:", data)
# 空轉
i = 0
while i< 100:
# children = zk.get_children('/node1',watch=event_listener)
# print('node1 has %s children with names %s' % (len(children), children))
time.sleep(1)
zk.stop()
zk.close()
關于kazooClient連接配接狀态說明
LOST
CONNECTED
SUSPENDED
KazooClient用戶端執行個體剛建立時，處于LOST狀态，同zookeeper建立連接配接後，轉為CONNECTED 。如果連接配接出問題、切換到不同的zookeeper叢集幾點，轉為SUSPENDED狀态，當你知道暫時不能執行指令，如果zookeeper節點不再是叢集的一部分，連接配接将丢失，也會導緻 SUSPENDED狀态
用戶端再次同zookeeper建立連接配接，如果會話不存在，用戶端連接配接狀态将轉為LOST，如果會話沒有過期，可用則轉為CONNECTED
運作效果
參考連結：
https://kazoo.readthedocs.io/en/latest/basic_usage.html
作者：授客
QQ：1033553122
全國軟體測試QQ交流群：7156436
Git位址：https://gitee.com/ishouke
友情提示：限于時間倉促，文中可能存在錯誤，歡迎指正、評論！
作者五行缺錢，如果覺得文章對您有幫助，請掃描下邊的二維碼打賞作者，金額随意，您的支援将是我繼續創作的源動力，打賞後如有任何疑問，請聯系我!!!
微信打賞
支付寶打賞 全國軟體測試交流QQ群