标签：des style blog http io color os ar for
问题导读：
1.本文三个角色之间是什么关系？
2.三个角色的作用是什么？
3.如何代码实现这三个角色的作用？
在 zookeeper适用场景：zookeeper解决了哪些问题有关于分布式集群配置文件同步问题的描述，本文介绍如何把zk应用到配置文件分发的场景。
假设有三个角色
配置文件同步到: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上
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发送数据包，格式如下
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
标签：des style blog http io color os ar for
原文地址：http://www.cnblogs.com/likehua/p/4060312.html
