Python通过zookeeper实现分布式服务代码解析
时间:2022-07-27
本文章向大家介绍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)
以上就是本文的全部内容,希望对大家的学习有所帮助。
- linux学习第三十六篇:MariaDB安装,Apache安装
- Python 5种连接字符串的方法
- Python 装饰器(Decorator)
- linux学习第三十七篇:安装PHP5,安装PHP7
- linux学习第三十八篇: Apache和PHP结合,Apache默认虚拟主机
- Python lambda表达式
- Python的闭包(Closure)与惰性计算(Lazy Evaluation)
- linux学习第三十九篇:Apache用户认证,域名跳转,Apache访问日志
- Python sorted()函数
- Python filter()函数
- Python reduce()函数
- Python map()函数
- Python iterator迭代器
- How To Implement The Decision Tree Algorithm From Scratch In Python (从零开始在Python中实现决策树算法)
- JavaScript 教程
- JavaScript 编辑工具
- JavaScript 与HTML
- JavaScript 与Java
- JavaScript 数据结构
- JavaScript 基本数据类型
- JavaScript 特殊数据类型
- JavaScript 运算符
- JavaScript typeof 运算符
- JavaScript 表达式
- JavaScript 类型转换
- JavaScript 基本语法
- JavaScript 注释
- Javascript 基本处理流程
- Javascript 选择结构
- Javascript if 语句
- Javascript if 语句的嵌套
- Javascript switch 语句
- Javascript 循环结构
- Javascript 循环结构实例
- Javascript 跳转语句
- Javascript 控制语句总结
- Javascript 函数介绍
- Javascript 函数的定义
- Javascript 函数调用
- Javascript 几种特殊的函数
- JavaScript 内置函数简介
- Javascript eval() 函数
- Javascript isFinite() 函数
- Javascript isNaN() 函数
- parseInt() 与 parseFloat()
- escape() 与 unescape()
- Javascript 字符串介绍
- Javascript length属性
- javascript 字符串函数
- Javascript 日期对象简介
- Javascript 日期对象用途
- Date 对象属性和方法
- Javascript 数组是什么
- Javascript 创建数组
- Javascript 数组赋值与取值
- Javascript 数组属性和方法
- Linux常见问题解决方案汇总
- Linux中文件/文件夹无法删除的解决方案
- GNU Parallel的具体使用
- crontab定时任务不执行的原因分析与解决方法
- 在Linux中安装ASPNET.Core3.0运行时的示例代码
- linux Bash脚本判别使用者的身份方法示例
- 配置Linux使用LDAP用户认证的方法
- linux Dig命令使用大全
- Ubuntu虚拟机下使用cutecom进行串口通信的方法
- 虚拟机Linux系统忘记密码修改root或其他用户密码的方法
- 在 RHEL8 /CentOS8 上建立多节点 Elastic stack 集群的方法
- linux 搭建svn服务器的方法步骤
- linux 下隐藏进程的一种方法及遇到的坑
- Ubuntu 18.04中截图工具shutter的编辑按钮不可用的解决办法
- Linux服务器利用防火墙iptables策略进行端口跳转的方法