思享工具箱

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

摘要：
问题指南：1。本文中三个角色之间的关系是什么？在ZooKeeper的应用场景中：ZooKeep解决了哪些问题？这里描述了分布式集群配置文件的同步问题。本文描述了如何将ZK应用于配置文件的分发场景。

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

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



在 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适用场景：配置文件同步第2张 


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

  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

　　

标签：#zookeeper集群#socket#try#zk | 浏览：168 | 发布日期：

免责声明：文章转载自《zookeeper适用场景：配置文件同步》仅用于学习参考。如对内容有疑问，请及时联系本站处理。

上篇php多版本使用composerfastdfs分布式文件系统部署下篇

宿迁高防，2C2G15M，22元/月；香港BGP，2C5G5M，25元/月 雨云优惠码：MjYwNzM=

相关文章

在Express和Socket.IO中使用session

在Express和Socket.IO中使用session_三水清 在Express和Socket.IO中使用session 在nodejs项目中对于一些认证需要用到session，例如我写的nodejs 聊天室的demo，就是通过session实现的认证。当存在session，直接进入聊天室，而不会重新登录。 在网上也找到不少关于Express框架中的...

Socket详解（转）

      在客户/服务器通信模式中, 客户端需要主动创建与服务器连接的 Socket(套接字), 服务器端收到了客户端的连接请求, 也会创建与客户连接的 Socket. Socket可看做是通信连接两端的收发器, 服务器与客户端都通过 Socket 来收发数据. 这篇文章首先介绍Socket类的各个构造方法, 以及成员方法的用法, 接着介绍 Sock...

zookeeper集群的python代码测试

上一篇已经讲解了如何安装zookeeper的python客户端，接下来是我在网上搜到的例子，举例应用环境是： 1.当有两个或者多个服务运行，并且同意时间只有一个服务接受请求(工作)，其他服务待命。 2.当接受请求(工作)的服务异常挂掉时，会从剩下的待命服务中选举出一个服务来接受请求(工作)。 下面直接上例子，有两个文件组成1.zkclient.py  ...

docker中zookeeper集群的安装与使用 以及zookeeper集群对mysq集群的管理

未完待续 在zookeeper容器中 配置文件的位置 /conf/zoo.cfg bash-4.4# cd /conf/bash-4.4# pwd /conf bash-4.4# ls configuration.xsl log4j.properties zoo.cfg zoo_sample.cfg bash-4.4# ls -...

vue+socket实现消息推送

前提：后台已设定好socket消息 首先在vue项目中引入socket。在npm下载socket。 npm install vue-socket.io 当然也可以在index.html中直接插入下面这句，但是最好不要这样做。 <script src='https://cdn.bootcss.com/socket.io/2.0.3/socket.io....

【java开发系列】—— java输入输出流

前言 　　任何语言输入输出流都是很重要的部分，比如从一个文件读入内容，进行分析，或者输出到另一个文件等等，都需要文件流的操作。这里简单介绍下reader,wirter,inputstream,outputstream的使用方法。其实Apache commons里面有个方法IOUtils可是实现方便快捷的流拷贝，感兴趣的可以参考官方文档。 　　JAVA的...

最新文章

随机推荐

思享工具箱导航