1. 云图网首页
  2. 技术专区
  3. 大数据

zookeeper适用场景：配置文件同步详解大数据

• 大数据

问题导读：
1.本文三个角色之间是什么关系？
2.三个角色的作用是什么？
3.如何代码实现这三个角色的作用？

zookeeper适用场景：配置文件同步详解大数据

在 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>zookeeper适用场景：配置文件同步详解大数据 

配置文件同步到: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 : [email protected] 
# 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 : [email protected] 
# 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发送数据包，格式如下

  1. 指令|数据
  2. 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

　　

原创文章，作者：Maggie-Hunter，如若转载，请注明出处：https://blog.ytso.com/9579.html

赞 (0)
0
上一篇 2021年7月19日
下一篇 2021年7月19日

相关推荐

发表回复

登录后才能评论