Python通过zookeeper实现分布式服务代码解析_Python

Python通过zookeeper实现分布式服务代码解析

2020-07-22 17:47Mars.wang Python

这篇文章主要介绍了Python通过zookeeper实现分布式服务代码解析,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下

借助zookeeper可以实现服务器的注册与发现,有需求的时候调用zookeeper来发现可用的服务器,将任务均匀分配到各个服务器上去.

这样可以方便的随任务的繁重程度对服务器进行弹性扩容,客户端和服务端是非耦合的,也可以随时增加客户端.

zk_server.py

									import threading

									import json

									import socket

									import sys

									from kazoo.client import KazooClient

									# TCP服务端绑定端口开启监听，同时将自己注册到zk

									class ZKServer(object):

									  def __init__(self, host, port):

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

									    self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

									    self.host = host

									    self.port = port

									    self.sock.bind((host, port))

									    self.zk = None

									  def serve(self):

									    """

									    开始服务，每次获取得到一个信息，都新建一个线程处理

									    """

									    self.sock.listen(128)

									    self.register_zk()

									    print("开始监听")

									    while True:

									      conn, addr = self.sock.accept()

									      print("建立链接%s" % str(addr))

									      t = threading.Thread(target=self.handle, args=(conn, addr))

									      t.start()

									  # 具体的处理逻辑,只要接收到数据就立即投入工作，下次没有数据本次链接结束

									  def handle(self, conn, addr):

									    while True:

									      data=conn.recv(1024)

									      if not data or data.decode('utf-8') == 'exit':

									        break

									      print(data.decode('utf-8'))

									    conn.close()

									    print('My work is done!!!')

									  # 将自己注册到zk，临时节点，所以连接不能中断

									  def register_zk(self):

									    """

									    注册到zookeeper

									    """

									    self.zk = KazooClient(hosts='127.0.0.1:2181')

									    self.zk.start()

									    self.zk.ensure_path('/rpc') # 创建根节点

									    value = json.dumps({'host': self.host, 'port': self.port})

									    # 创建服务子节点

									    self.zk.create('/rpc/server', value.encode(), ephemeral=True, sequence=True)

									if __name__ == '__main__':

									  if len(sys.argv) < 3:

									    print("usage:python server.py [host] [port]")

									    exit(1)

									  host = sys.argv[1]

									  port = sys.argv[2]

									  server = ZKServer(host, int(port))

									  server.serve()

zk_client.py

									import random

									import sys

									import time

									import json

									import socket

									from kazoo.client import KazooClient

									# 客户端连接zk,并从zk获取可用的服务器列表

									class ZKClient(object):

									  def __init__(self):

									    self._zk = KazooClient(hosts='127.0.0.1:2181')

									    self._zk.start()

									    self._get_servers()

									  def _get_servers(self, event=None):

									    """

									    从zookeeper获取服务器地址信息列表

									    """

									    servers = self._zk.get_children('/rpc', watch=self._get_servers)

									    # print(servers)

									    self._servers = []

									    for server in servers:

									      data = self._zk.get('/rpc/' + server)[0]

									      if data:

									        addr = json.loads(data.decode())

									        self._servers.append(addr)

									  def _get_server(self):

									    """

									    随机选出一个可用的服务器

									    """

									    return random.choice(self._servers)

									  def get_connection(self):

									    """

									    提供一个可用的tcp连接

									    """

									    sock = None

									    while True:

									      server = self._get_server()

									      print('server:%s' % server)

									      try:

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

									        sock.connect((server['host'], server['port']))

									      except ConnectionRefusedError:

									        time.sleep(1)

									        continue

									      else:

									        break

									    return sock

									if __name__ == '__main__':

									  # 模拟多个客户端批量生成任务，推送给服务器执行

									  client = ZKClient()

									  for i in range(40):

									    sock = client.get_connection()

									    sock.send(bytes(str(i), encoding='utf8'))

									    sock.close()

									    time.sleep(1)