问题导读：
1.本文三个角色之间是什么关系？
2.三个角色的作用是什么？
3.如何代码实现这三个角色的作用？
在 zookeeper适用场景：zookeeper解决了哪些问题有关于分布式集群配置文件同步问题的描述，本文介绍如何把zk应用到配置文件分发的场景。
假设有三个角色
- 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
#!/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上
- 初始化连接到zk service
- 获取所有/Applications/NginxConf的znode列表并且设置Watcher
- 找到列表中最大的znode节点，记录它的编号，然后getData获取它的数据，这里就拿到了最新的配置信息
- 每次又Watcher被触发，就获取列表中编号最大的节点，然后把上一次保存的编号和它比较，一旦又更新就重新获取数据，应用到worker，如果相同就不需要到zk上获取数据。
vim zk_app.py
#!/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发送数据包，格式如下
- 指令|数据
- ADD|helloworld
vim trigger.py
#!/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()
运行
$ python zk_agent.py
$ python zk_app.py
$ python trigger.py