图文精华

About云-梭伦科技»专题 › 技术学习(版主发帖区) › 大数据学习 › ZooKeeper › 实操演练› zookeeper适用场景：配置文件同步
分享

zookeeper适用场景：配置文件同步

pig2 2014-9-23 16:05:28 发表于 实操演练 [显示全部楼层] 只看大图 回帖奖励 阅读模式 关闭右栏 1 46854
问题导读：
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

    zk_conf.png

配置文件同步到: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

  1. #!/usr/bin/env python2.7

  2. # -*- coding: UTF-8 -*-

  3.  

  4. # author : firefoxbug

  5. # E-Mail : wanghuafire@gmail.com

  6. # Blog   : www.firefoxbug.net

  7. 

  8. import logging

  9. from os.path import basename, join

  10. 

  11. from zkclient import ZKClient, zookeeper, watchmethod

  12. 

  13. import os

  14. import sys

  15. import threading

  16. import signal

  17. import time

  18. 

  19. logging.basicConfig(

  20.     level = logging.DEBUG,

  21.     format = "[%(asctime)s] %(levelname)-8s %(message)s"

  22. )

  23. 

  24. log = logging

  25. 

  26. class GJZookeeper(object):

  27. 

  28.     ZK_HOST = "localhost:2181"

  29.     ROOT = "/Roles"

  30.     WORKERS_PATH = join(ROOT, "workers")

  31.     MASTERS_NUM = 1

  32.     TIMEOUT = 10000

  33. 

  34.     def __init__(self, verbose = True):

  35.         self.VERBOSE = verbose

  36.         self.masters = []

  37.         self.is_master = False

  38.         self.path = None

  39. 

  40.         self.APP_ROOT = "/Applications"

  41.         self.APP_CONF = join(self.APP_ROOT,"NginxConf")

  42. 

  43.         self.zk = ZKClient(self.ZK_HOST, timeout = self.TIMEOUT)

  44.         self.say("login zookeeper successfully!")

  45.         # init

  46.         self.create_roles_znode()

  47.         # register

  48.         self.register()

  49. 

  50.     def create_roles_znode(self):

  51.         """

  52.         create the zookeeper node if not exist

  53.         |-Roles

  54.              |-workers

  55.         """

  56.         nodes = (self.ROOT, self.WORKERS_PATH)

  57.         for node in nodes: 

  58.             if not self.zk.exists(node):

  59.                 try:

  60.                     self.zk.create(node, "")

  61.                 except:

  62.                     pass

  63. 

  64.     @property

  65.     def is_slave(self):

  66.         return not self.is_master

  67. 

  68.     def register(self):

  69.         """

  70.         register a node for this worker,znode type : EPHEMERAL | SEQUENCE

  71.         |-Roles

  72.              |-workers

  73.                      |-worker000000000x         ==>>master

  74.                      |-worker000000000x+1       ==>>worker

  75.                      ....

  76.         """

  77.         self.path = self.zk.create(self.WORKERS_PATH + "/worker", "1", flags=zookeeper.EPHEMERAL | zookeeper.SEQUENCE)

  78.         self.path = basename(self.path)

  79.         self.say("I'm %s" % self.path)

  80.         # check who is the master

  81.         self.get_master()

  82. 

  83.     def get_master(self):

  84.         """

  85.         get children, and check who is the smallest child

  86.         """

  87.         @watchmethod

  88.         def watcher(event):

  89.             self.say("child changed, try to get master again.")

  90.             self.get_master()

  91. 

  92.         children = self.zk.get_children(self.WORKERS_PATH, watcher)

  93.         children.sort()

  94.         self.say("%s's children: %s" % (self.WORKERS_PATH, children)) 

  95. 

  96.         # check if I'm master

  97.         self.masters = children[:self.MASTERS_NUM]

  98.         if self.path in self.masters:

  99.             self.is_master = True

  100.             self.say("I've become master!")

  101.             self.create_app_znode()

  102.         else:

  103.             self.say("%s is masters, I'm slave" % self.masters)

  104. 

  105.     def create_app_znode(self):

  106.         """

  107.         create the zookeeper node if not exist

  108.         |-Applications

  109.                     |-NginxConf

  110.         """

  111.         nodes = (self.APP_ROOT, self.APP_CONF)

  112.         for node in nodes: 

  113.             if not self.zk.exists(node):

  114.                 try:

  115.                     self.say("Create znode [%s] ..."%(node))

  116.                     self.zk.create(node, "")

  117.                 except:

  118.                     pass

  119. 

  120.     def create_conf_znode(self,data):

  121.         """

  122.         create the zookeeper node's children if not exist,contents is conf data

  123.         |-Applications

  124.                     |-NginxConf

  125.                             |-item-000000000x => data

  126.         """

  127.         self.child_node = join(self.APP_CONF,"conf-")

  128.         path = self.zk.create(self.child_node,data, flags=zookeeper.SEQUENCE|zookeeper.PERSISTENT)

  129.         self.say("create znode %s"%path)

  130. 

  131.     def say(self, msg):

  132.         """

  133.         print messages to screen

  134.         """

  135.         if self.VERBOSE:

  136.             if self.path:

  137.                 log.info("[ %s(%s) ] %s" % (self.path, "master" if self.is_master else "slave", msg))

  138.             else:

  139.                 log.info(msg)

  140.         

  141. class Watcher:

  142.     def __init__(self):

  143.         """ Creates a child thread, which returns.  The parent

  144.             thread waits for a KeyboardInterrupt and then kills

  145.             the child thread.

  146.         """

  147.         self.child = os.fork()

  148.         if self.child == 0:

  149.             return

  150.         else:

  151.             self.watch()

  152. 

  153.     def watch(self):

  154.         try:

  155.             os.wait()

  156.         except KeyboardInterrupt:

  157.             print ' exit...'

  158.             self.kill()

  159.         sys.exit()

  160. 

  161.     def kill(self):

  162.         try:

  163.             os.kill(self.child, signal.SIGKILL)

  164.         except OSError:

  165.             pass

  166. 

  167. 

  168. def start_zk_worker():

  169.     """

  170.         连接到zookeeper执行初始化

  171.     """

  172.     gj_zookeeper = GJZookeeper()

  173.     th1 = threading.Thread(target = start_agent_worker, name = "thread_1", args = (gj_zookeeper,))

  174. 

  175.     th1.start()

  176. 

  177.     th1.join()

  178. 

  179. def start_agent_worker(gj_zookeeper):

  180.     """

  181.         监听配置文件变更信息，解析指令

  182.     """

  183.     import socket

  184. 

  185.     address = ('', 8877)

  186.     s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # s = socket.socket()

  187.     s.bind(address)

  188.     s.listen(5)

  189.     print "listen on port 8877 ..."

  190. 

  191.     while True:

  192.         ss, addr = s.accept()

  193.         print "receive connetcion from " ,addr

  194.         content = ""

  195.         while True:

  196.             try :

  197.                 data = ss.recv(512)

  198.                 if not data:

  199.                     print "close connetcion " ,addr

  200.                     ss.close()

  201.                     break

  202.                 content = content + data

  203.                 print "receive message from %s : %s"%(addr[0],data)

  204.             except Exception,e:

  205.                 print "receive error from %s : %s"%(addr[0],str(e))

  206.                 ss.close()

  207.                 break   

  208.         parse_trigger_package(content,gj_zookeeper)

  209. 

  210.     s.close()

  211. 

  212. def parse_trigger_package(data,gj_zookeeper):

  213.     try:

  214.         cmd = data.split('|')[0]

  215.         content = data.split('|')[1]

  216.     except Exception, e:

  217.         print "ERROR :",str(e)

  218.         return 

  219.     if cmd == "ADD":

  220.         gj_zookeeper.create_conf_znode(content)

  221.     else:

  222.         pass

  223. 

  224. 

  225. def main():

  226.     Watcher()

  227.     start_zk_worker()

  228. 

  229. if __name__ == "__main__":

  230.     main()

  231.     import time

  232.     time.sleep(20000)
复制代码


配置文件接收应用:zk_appzk_app逻辑如下，它工作在每台worker上
  • 初始化连接到zk service
  • 获取所有/Applications/NginxConf的znode列表并且设置Watcher
  • 找到列表中最大的znode节点，记录它的编号，然后getData获取它的数据，这里就拿到了最新的配置信息
  • 每次又Watcher被触发，就获取列表中编号最大的节点，然后把上一次保存的编号和它比较，一旦又更新就重新获取数据，应用到worker，如果相同就不需要到zk上获取数据。
vim zk_app.py

  1. #!/usr/bin/env python2.7

  2. # -*- coding: UTF-8 -*-

  3.  

  4. # author : firefoxbug

  5. # E-Mail : wanghuafire@gmail.com

  6. # Blog   : www.firefoxbug.net

  7. 

  8. import logging

  9. from os.path import basename, join

  10. 

  11. from zkclient import ZKClient, zookeeper, watchmethod

  12. 

  13. import os

  14. import sys

  15. import threading

  16. import signal

  17. import time

  18. 

  19. logging.basicConfig(

  20.     level = logging.DEBUG,

  21.     format = "[%(asctime)s] %(levelname)-8s %(message)s"

  22. )

  23. 

  24. log = logging

  25. 

  26. class GJZookeeper(object):

  27. 

  28.     ZK_HOST = "localhost:2181"

  29.     ROOT = "/Roles"

  30.     WORKERS_PATH = join(ROOT, "workers")

  31.     MASTERS_NUM = 1

  32.     TIMEOUT = 10000

  33. 

  34.     def __init__(self, verbose = True):

  35.         self.VERBOSE = verbose

  36.         self.masters = []

  37.         self.is_master = False

  38.         self.path = None

  39. 

  40.         self.APP_ROOT = "/Applications"

  41.         self.APP_CONF = join(self.APP_ROOT,"NginxConf")

  42. 

  43.         self.zk = ZKClient(self.ZK_HOST, timeout = self.TIMEOUT)

  44.         self.say("login zookeeper successfully!")

  45.         # init

  46.         self.create_roles_znode()

  47.         # register

  48.         self.register()

  49. 

  50.     def create_roles_znode(self):

  51.         """

  52.         create the zookeeper node if not exist

  53.         |-Roles

  54.              |-workers

  55.         """

  56.         nodes = (self.ROOT, self.WORKERS_PATH)

  57.         for node in nodes: 

  58.             if not self.zk.exists(node):

  59.                 try:

  60.                     self.zk.create(node, "")

  61.                 except:

  62.                     pass

  63. 

  64.     @property

  65.     def is_slave(self):

  66.         return not self.is_master

  67. 

  68.     def register(self):

  69.         """

  70.         register a node for this worker,znode type : EPHEMERAL | SEQUENCE

  71.         |-Roles

  72.              |-workers

  73.                      |-worker000000000x         ==>>master

  74.                      |-worker000000000x+1       ==>>worker

  75.                      ....

  76.         """

  77.         self.path = self.zk.create(self.WORKERS_PATH + "/worker", "1", flags=zookeeper.EPHEMERAL | zookeeper.SEQUENCE)

  78.         self.path = basename(self.path)

  79.         self.say("I'm %s" % self.path)

  80.         # check who is the master

  81.         self.get_master()

  82. 

  83.     def get_master(self):

  84.         """

  85.         get children, and check who is the smallest child

  86.         """

  87.         @watchmethod

  88.         def watcher(event):

  89.             self.say("child changed, try to get master again.")

  90.             self.get_master()

  91. 

  92.         children = self.zk.get_children(self.WORKERS_PATH, watcher)

  93.         children.sort()

  94.         self.say("%s's children: %s" % (self.WORKERS_PATH, children)) 

  95. 

  96.         # check if I'm master

  97.         self.masters = children[:self.MASTERS_NUM]

  98.         if self.path in self.masters:

  99.             self.is_master = True

  100.             self.say("I've become master!")

  101.             self.create_app_znode()

  102.         else:

  103.             self.say("%s is masters, I'm slave" % self.masters)

  104. 

  105.     def create_app_znode(self):

  106.         """

  107.         create the zookeeper node if not exist

  108.         |-Applications

  109.                     |-NginxConf

  110.         """

  111.         nodes = (self.APP_ROOT, self.APP_CONF)

  112.         for node in nodes: 

  113.             if not self.zk.exists(node):

  114.                 try:

  115.                     self.say("Create znode [%s] ..."%(node))

  116.                     self.zk.create(node, "")

  117.                 except:

  118.                     pass

  119. 

  120.     def create_conf_znode(self,data):

  121.         """

  122.         create the zookeeper node's children if not exist,contents is conf data

  123.         |-Applications

  124.                     |-NginxConf

  125.                             |-conf-000000000x => data

  126.         """

  127.         self.child_node = join(self.APP_CONF,"conf-")

  128.         path = self.zk.create(self.child_node,data, flags=zookeeper.SEQUENCE|zookeeper.PERSISTENT)

  129.         self.say("create znode %s"%path)

  130. 

  131.     def say(self, msg):

  132.         """

  133.         print messages to screen

  134.         """

  135.         if self.VERBOSE:

  136.             if self.path:

  137.                 log.info("[ %s(%s) ] %s" % (self.path, "master" if self.is_master else "slave", msg))

  138.             else:

  139.                 log.info(msg)

  140.         

  141. class Watcher:

  142.     def __init__(self):

  143.         """ Creates a child thread, which returns.  The parent

  144.             thread waits for a KeyboardInterrupt and then kills

  145.             the child thread.

  146.         """

  147.         self.child = os.fork()

  148.         if self.child == 0:

  149.             return

  150.         else:

  151.             self.watch()

  152. 

  153.     def watch(self):

  154.         try:

  155.             os.wait()

  156.         except KeyboardInterrupt:

  157.             print ' exit...'

  158.             self.kill()

  159.         sys.exit()

  160. 

  161.     def kill(self):

  162.         try:

  163.             os.kill(self.child, signal.SIGKILL)

  164.         except OSError:

  165.             pass

  166. 

  167. 

  168. def start_zk_worker():

  169.     """

  170.         连接到zookeeper执行初始化

  171.     """

  172.     gj_zookeeper = GJZookeeper()

  173.     th1 = threading.Thread(target = start_agent_worker, name = "thread_1", args = (gj_zookeeper,))

  174. 

  175.     th1.start()

  176. 

  177.     th1.join()

  178. 

  179. def start_agent_worker(gj_zookeeper):

  180.     """

  181.         监听配置文件变更信息，解析指令

  182.     """

  183.     import socket

  184. 

  185.     address = ('', 8877)

  186.     s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # s = socket.socket()

  187.     s.bind(address)

  188.     s.listen(5)

  189.     print "listen on port 8877 ..."

  190. 

  191.     while True:

  192.         ss, addr = s.accept()

  193.         print "receive connetcion from " ,addr

  194.         content = ""

  195.         while True:

  196.             try :

  197.                 data = ss.recv(512)

  198.                 if not data:

  199.                     print "close connetcion " ,addr

  200.                     ss.close()

  201.                     break

  202.                 content = content + data

  203.                 print "receive message from %s : %s"%(addr[0],data)

  204.             except Exception,e:

  205.                 print "receive error from %s : %s"%(addr[0],str(e))

  206.                 ss.close()

  207.                 break   

  208.         parse_trigger_package(content,gj_zookeeper)

  209. 

  210.     s.close()

  211. 

  212. def parse_trigger_package(data,gj_zookeeper):

  213.     try:

  214.         cmd = data.split('|')[0]

  215.         content = data.split('|')[1]

  216.     except Exception, e:

  217.         print "ERROR :",str(e)

  218.         return 

  219.     if cmd == "ADD":

  220.         gj_zookeeper.create_conf_znode(content)

  221.     else:

  222.         pass

  223. 

  224. 

  225. def main():

  226.     Watcher()

  227.     start_zk_worker()

  228. 

  229. if __name__ == "__main__":

  230.     main()

  231.     import time

  232.     time.sleep(20000)
复制代码


配置文件发送:trigger实现的逻辑很简单，就是模拟给zk_agent发送数据包，格式如下

  1. 指令|数据

  2. ADD|helloworld
复制代码

vim trigger.py

  1. #!/usr/bin/python

  2. import socket  

  3. import sys

  4. 

  5. sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)  

  6. sock.connect(('127.0.0.1', 8877))

  7. str = sys.argv[1] + '|' + sys.argv[2]

  8. sock.send(str)  

  9. sock.close()
复制代码


运行
  1. $ python zk_agent.py

  2. $ python zk_app.py

  3. $ python trigger.py
复制代码



相关文章：
zookeeper原理
zookeeper中Watcher和Notifications
zookeeper适用场景：如何竞选Master及代码实现
zookeeper适用场景：配置文件同步
zookeeper适用场景：分布式锁实现
zookeeper适用场景：zookeeper解决了哪些问题





本帖被以下淘专辑推荐:

关注公众号，获取大数据、人工智能20套、区块链资源5阶段等资源，随时更新，获取最新技术资源
回复

使用道具 举报

自定义广告语

已有(1)人评论

跳转到指定楼层
正序浏览
李新魁 发表于 2014-10-11 15:10:58
$ python zk_agent.py
$ python zk_app.py
$ python trigger.py
一定要运行3个代码脚本才能实现配置文件的同步么？他不是一定时间内就会同步你修改过的配置文件么？
回复

使用道具 举报

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

发表新帖
pig2

超级版主

关注

  • 2332

    主题

  • 3342

    帖子

  • 1972

    粉丝

欢迎大家如about云官方2群322273151，更新咨询，更新资源，随时关注
TA的主题
关闭

推荐上一条 /2 下一条

关于我们· 联系我们· 加入我们· 小黑屋· 合作伙伴   

Copyright © 2001-2025 About云-梭伦科技 Powered by Discuz! X3.4 Licensed Discuz Team.

简书 / Smrz 京ICP备2020039040号 Wxb 简书网举报电话：021-34700000