zookeeper適用場景：配置文件同步
問題導讀：
1.本文三個角色之間是什麼關係？
2.三個角色的做用是什麼？
3.如何代碼實現這三個角色的做用？
在 zookeeper適用場景：zookeeper解決了哪些問題有關於分佈式集羣配置文件同步問題的描述，本文介紹如何把zk應用到配置文件分發的場景。
假設有三個角色php
- trigger：發佈最新的配置文件數據，發送指令和數據給zk_agent，實現是下面的trigger.py
- zk_agent：接收來自trigger.py觸發的指令和數據，而且把數據更新到zk service上，從而觸發zk_app來獲取最新的配置數據，實現是下面的zk_agent.py
- zk_app：部署在每臺worker上的註冊監聽zk中配置文件所在znode的變化，從而獲取最新的配置文件，應用到worker中，實現是下面的zk_app.py
<ignore_js_op>
配置文件同步到:zk_agent實現邏輯：
* 初始化鏈接到zk service，首先競選出master
* master create一個配置文件管理的PERSISTENT類型的znode，好比是/Applications/NginxConf，
* 啓動一個線程，專門接收trigger發送的指令，一收到指令，就create一個"conf-"開頭的SEQUENCE|PERSISTENT類型的節點，指定znode數據內容是從trigger收到的數據
* 第一次剛更新會建立節點/Applications/NginxConf/conf-0000000000,之後每次更新新的配置文件編號就會增大。
vim zk_agent.py
html
#!/usr/bin/env python2.7
# -*- coding: UTF-8 -*-
# author : firefoxbug
# E-Mail : wanghuafire@gmail.com
# Blog : www.firefoxbug.net
import logging
from os.path import basename, join
from zkclient import ZKClient, zookeeper, watchmethod
import os
import sys
import threading
import signal
import time
logging.basicConfig(
level = logging.DEBUG,
format = "[%(asctime)s] %(levelname)-8s %(message)s"
)
log = logging
class GJZookeeper(object):
ZK_HOST = "localhost:2181"
ROOT = "/Roles"
WORKERS_PATH = join(ROOT, "workers")
MASTERS_NUM = 1
TIMEOUT = 10000
def __init__(self, verbose = True):
self.VERBOSE = verbose
self.masters = []
self.is_master = False
self.path = None
self.APP_ROOT = "/Applications"
self.APP_CONF = join(self.APP_ROOT,"NginxConf")
self.zk = ZKClient(self.ZK_HOST, timeout = self.TIMEOUT)
self.say("login zookeeper successfully!")
# init
self.create_roles_znode()
# register
self.register()
def create_roles_znode(self):
"""
create the zookeeper node if not exist
|-Roles
|-workers
"""
nodes = (self.ROOT, self.WORKERS_PATH)
for node in nodes:
if not self.zk.exists(node):
try:
self.zk.create(node, "")
except:
pass
@property
def is_slave(self):
return not self.is_master
def register(self):
"""
register a node for this worker,znode type : EPHEMERAL | SEQUENCE
|-Roles
|-workers
|-worker000000000x ==>>master
|-worker000000000x+1 ==>>worker
....
"""
self.path = self.zk.create(self.WORKERS_PATH + "/worker", "1", flags=zookeeper.EPHEMERAL | zookeeper.SEQUENCE)
self.path = basename(self.path)
self.say("I'm %s" % self.path)
# check who is the master
self.get_master()
def get_master(self):
"""
get children, and check who is the smallest child
"""
@watchmethod
def watcher(event):
self.say("child changed, try to get master again.")
self.get_master()
children = self.zk.get_children(self.WORKERS_PATH, watcher)
children.sort()
self.say("%s's children: %s" % (self.WORKERS_PATH, children))
# check if I'm master
self.masters = children[:self.MASTERS_NUM]
if self.path in self.masters:
self.is_master = True
self.say("I've become master!")
self.create_app_znode()
else:
self.say("%s is masters, I'm slave" % self.masters)
def create_app_znode(self):
"""
create the zookeeper node if not exist
|-Applications
|-NginxConf
"""
nodes = (self.APP_ROOT, self.APP_CONF)
for node in nodes:
if not self.zk.exists(node):
try:
self.say("Create znode [%s] ..."%(node))
self.zk.create(node, "")
except:
pass
def create_conf_znode(self,data):
"""
create the zookeeper node's children if not exist,contents is conf data
|-Applications
|-NginxConf
|-item-000000000x => data
"""
self.child_node = join(self.APP_CONF,"conf-")
path = self.zk.create(self.child_node,data, flags=zookeeper.SEQUENCE|zookeeper.PERSISTENT)
self.say("create znode %s"%path)
def say(self, msg):
"""
print messages to screen
"""
if self.VERBOSE:
if self.path:
log.info("[ %s(%s) ] %s" % (self.path, "master" if self.is_master else "slave", msg))
else:
log.info(msg)
class Watcher:
def __init__(self):
""" Creates a child thread, which returns. The parent
thread waits for a KeyboardInterrupt and then kills
the child thread.
"""
self.child = os.fork()
if self.child == 0:
return
else:
self.watch()
def watch(self):
try:
os.wait()
except KeyboardInterrupt:
print ' exit...'
self.kill()
sys.exit()
def kill(self):
try:
os.kill(self.child, signal.SIGKILL)
except OSError:
pass
def start_zk_worker():
"""
鏈接到zookeeper執行初始化
"""
gj_zookeeper = GJZookeeper()
th1 = threading.Thread(target = start_agent_worker, name = "thread_1", args = (gj_zookeeper,))
th1.start()
th1.join()
def start_agent_worker(gj_zookeeper):
"""
監聽配置文件變動信息，解析指令
"""
import socket
address = ('', 8877)
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # s = socket.socket()
s.bind(address)
s.listen(5)
print "listen on port 8877 ..."
while True:
ss, addr = s.accept()
print "receive connetcion from " ,addr
content = ""
while True:
try :
data = ss.recv(512)
if not data:
print "close connetcion " ,addr
ss.close()
break
content = content + data
print "receive message from %s : %s"%(addr[0],data)
except Exception,e:
print "receive error from %s : %s"%(addr[0],str(e))
ss.close()
break
parse_trigger_package(content,gj_zookeeper)
s.close()
def parse_trigger_package(data,gj_zookeeper):
try:
cmd = data.split('|')[0]
content = data.split('|')[1]
except Exception, e:
print "ERROR :",str(e)
return
if cmd == "ADD":
gj_zookeeper.create_conf_znode(content)
else:
pass
def main():
Watcher()
start_zk_worker()
if __name__ == "__main__":
main()
import time
time.sleep(20000)
配置文件接收應用:zk_appzk_app邏輯以下，它工做在每臺worker上
node
- 初始化鏈接到zk service
- 獲取全部/Applications/NginxConf的znode列表而且設置Watcher
- 找到列表中最大的znode節點，記錄它的編號，而後getData獲取它的數據，這裏就拿到了最新的配置信息
- 每次又Watcher被觸發，就獲取列表中編號最大的節點，而後把上一次保存的編號和它比較，一旦又更新就從新獲取數據，應用到worker，若是相同就不須要到zk上獲取數據。
vim zk_app.py
python
#!/usr/bin/env python2.7
# -*- coding: UTF-8 -*-
# author : firefoxbug
# E-Mail : wanghuafire@gmail.com
# Blog : www.firefoxbug.net
import logging
from os.path import basename, join
from zkclient import ZKClient, zookeeper, watchmethod
import os
import sys
import threading
import signal
import time
logging.basicConfig(
level = logging.DEBUG,
format = "[%(asctime)s] %(levelname)-8s %(message)s"
)
log = logging
class GJZookeeper(object):
ZK_HOST = "localhost:2181"
ROOT = "/Roles"
WORKERS_PATH = join(ROOT, "workers")
MASTERS_NUM = 1
TIMEOUT = 10000
def __init__(self, verbose = True):
self.VERBOSE = verbose
self.masters = []
self.is_master = False
self.path = None
self.APP_ROOT = "/Applications"
self.APP_CONF = join(self.APP_ROOT,"NginxConf")
self.zk = ZKClient(self.ZK_HOST, timeout = self.TIMEOUT)
self.say("login zookeeper successfully!")
# init
self.create_roles_znode()
# register
self.register()
def create_roles_znode(self):
"""
create the zookeeper node if not exist
|-Roles
|-workers
"""
nodes = (self.ROOT, self.WORKERS_PATH)
for node in nodes:
if not self.zk.exists(node):
try:
self.zk.create(node, "")
except:
pass
@property
def is_slave(self):
return not self.is_master
def register(self):
"""
register a node for this worker,znode type : EPHEMERAL | SEQUENCE
|-Roles
|-workers
|-worker000000000x ==>>master
|-worker000000000x+1 ==>>worker
....
"""
self.path = self.zk.create(self.WORKERS_PATH + "/worker", "1", flags=zookeeper.EPHEMERAL | zookeeper.SEQUENCE)
self.path = basename(self.path)
self.say("I'm %s" % self.path)
# check who is the master
self.get_master()
def get_master(self):
"""
get children, and check who is the smallest child
"""
@watchmethod
def watcher(event):
self.say("child changed, try to get master again.")
self.get_master()
children = self.zk.get_children(self.WORKERS_PATH, watcher)
children.sort()
self.say("%s's children: %s" % (self.WORKERS_PATH, children))
# check if I'm master
self.masters = children[:self.MASTERS_NUM]
if self.path in self.masters:
self.is_master = True
self.say("I've become master!")
self.create_app_znode()
else:
self.say("%s is masters, I'm slave" % self.masters)
def create_app_znode(self):
"""
create the zookeeper node if not exist
|-Applications
|-NginxConf
"""
nodes = (self.APP_ROOT, self.APP_CONF)
for node in nodes:
if not self.zk.exists(node):
try:
self.say("Create znode [%s] ..."%(node))
self.zk.create(node, "")
except:
pass
def create_conf_znode(self,data):
"""
create the zookeeper node's children if not exist,contents is conf data
|-Applications
|-NginxConf
|-conf-000000000x => data
"""
self.child_node = join(self.APP_CONF,"conf-")
path = self.zk.create(self.child_node,data, flags=zookeeper.SEQUENCE|zookeeper.PERSISTENT)
self.say("create znode %s"%path)
def say(self, msg):
"""
print messages to screen
"""
if self.VERBOSE:
if self.path:
log.info("[ %s(%s) ] %s" % (self.path, "master" if self.is_master else "slave", msg))
else:
log.info(msg)
class Watcher:
def __init__(self):
""" Creates a child thread, which returns. The parent
thread waits for a KeyboardInterrupt and then kills
the child thread.
"""
self.child = os.fork()
if self.child == 0:
return
else:
self.watch()
def watch(self):
try:
os.wait()
except KeyboardInterrupt:
print ' exit...'
self.kill()
sys.exit()
def kill(self):
try:
os.kill(self.child, signal.SIGKILL)
except OSError:
pass
def start_zk_worker():
"""
鏈接到zookeeper執行初始化
"""
gj_zookeeper = GJZookeeper()
th1 = threading.Thread(target = start_agent_worker, name = "thread_1", args = (gj_zookeeper,))
th1.start()
th1.join()
def start_agent_worker(gj_zookeeper):
"""
監聽配置文件變動信息，解析指令
"""
import socket
address = ('', 8877)
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # s = socket.socket()
s.bind(address)
s.listen(5)
print "listen on port 8877 ..."
while True:
ss, addr = s.accept()
print "receive connetcion from " ,addr
content = ""
while True:
try :
data = ss.recv(512)
if not data:
print "close connetcion " ,addr
ss.close()
break
content = content + data
print "receive message from %s : %s"%(addr[0],data)
except Exception,e:
print "receive error from %s : %s"%(addr[0],str(e))
ss.close()
break
parse_trigger_package(content,gj_zookeeper)
s.close()
def parse_trigger_package(data,gj_zookeeper):
try:
cmd = data.split('|')[0]
content = data.split('|')[1]
except Exception, e:
print "ERROR :",str(e)
return
if cmd == "ADD":
gj_zookeeper.create_conf_znode(content)
else:
pass
def main():
Watcher()
start_zk_worker()
if __name__ == "__main__":
main()
import time
time.sleep(20000)
配置文件發送:trigger實現的邏輯很簡單，就是模擬給zk_agent發送數據包，格式以下
vim
- 指令|數據
- ADD|helloworld
vim trigger.py
app
#!/usr/bin/python
import socket
import sys
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(('127.0.0.1', 8877))
str = sys.argv[1] + '|' + sys.argv[2]
sock.send(str)
sock.close()
運行python2.7
$ python zk_agent.py $ python zk_app.py $ python trigger.py
相關文章
- 1. kickstart配置文件適配 vlan 場景應用
- 2. 使用zookeeper實現配置同步
- 3. 使用ZooKeeper實現配置同步
- 4. zookeeper(三)-zookeeper應用場景
- 5. ZooKeeper應用場景
- 6. zookeeper應用場景
- 7. Zookeeper 應用場景
- 8. Zookeeper 使用場景
- 9. zookeeper使用場景
- 10. Zookeeper配置文件
- 更多相關文章...
- • Maven 構建配置文件 - Maven教程
- • MyBatis配置文件詳解 - MyBatis教程
- • IDEA下SpringBoot工程配置文件沒有提示
- • IntelliJ IDEA 代碼格式化配置和快捷鍵
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
歡迎關注本站公眾號,獲取更多信息
相關文章