基于Python及zookeeper實作簡單分布式任務排程系統設計思路及核心代碼實作
by:授客 QQ：1033553122
測試環境
功能需求
實作思路
代碼實踐(關鍵技術點實作)
代碼子產品組織結構
配置檔案解析
MyTCPServer.py
MyTCPClient.py
appClient.py
loadAgent.py
運作效果 13
測試環境
Win7 64位
Linux 64位
Python 3.3.4
kazoo-2.6.1-py2.py3-none-any.whl（windows）
kazoo-2.6.1.tar.gz (linux)
https://pypi.org/project/kazoo/#files
zookeeper-3.4.13.tar.gz
下載下傳位址1：
http://zookeeper.apache.org/releases.html#download
https://www.apache.org/dyn/closer.cgi/zookeeper/
https://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper/
功能需求
把不同的負載主機，注冊為zookeeper的節點，其它應用子產品請求zookeeper擷取相關節點資訊(伺服器ip，端口号，伺服器任務執行狀态)，通過伺服器任務狀态選擇沒有運作指定任務的伺服器執行相關任務。
針對以上需求，做一個技術預研，核心代碼實作
實作思路
負載伺服器啟動時，初始化zookeeper用戶端，建立tcp伺服器，注冊節點資訊到zookeeper伺服器(資訊包含tcp伺服器ip，端口，負載伺服器任務執行狀态)，然後定時檢測負載伺服器任務執行狀态（通過檢測某個程序的名稱是否存在進行判斷），其它應用子產品通過zookeeper擷取節點資訊後，通過tcp socket通信，向負載伺服器發送執行指令，然後負載伺服器根據這些指令進行不同的處理。
代碼實踐(關鍵技術點實作)
代碼子產品組織結構
配置檔案解析
conf/agent.conf
[AGENT]
interval = 5
proc = sftp-server
[README]
interval = 更新伺服器節點資訊頻率（機關 秒
proc = 需要檢測的程序名稱(程式通過查找對應程序名稱來判斷負載程式是否還在運作，進而判斷伺服器狀态
conf/tcpserver.conf
[TCPSERVER]
host=10.202.7.165
port = 8000
host = tcp伺服器主機位址
port = tcp伺服器監聽端口
conf/zookeeper.conf
[ZOOKEEPER]
hosts = 10.118.52.26:2181
nodeParentPath=/rootNode
hosts = zookeeper位址，如果是叢集位址，即有多個，用英文逗号分隔
nodeParentPath=負載機節點所在父級路徑
MyTCPServer.py
#!/usr/bin/env python 3.4.0
#-*- encoding:utf-8 -*-
__author__ = 'shouke'
import socketserver
from log import logger
class MyTCPHandler(socketserver.BaseRequestHandler):
"""
The RequestHandler class for our server.
It is instantiated once per connection to the server, and must
override the handle() method to implement communication to the
client.
def handle(self):
while True:
# self.request is the TCP socket connected to the client
self.data = self.request.recv(1024).decode('utf-8').strip()
logger.info('receive data from client[host：%s port：%s]：%s' % (self.client_address[0], self.client_address[1], self.data))
if self.data == 'bye':
self.request.sendall(bytes('bye', encoding='utf-8'))
self.request.close()
break
else:
self.request.sendall(self.data.upper().encode('utf-8'))
class MyTCPServer:
def __init__(self, host, port):
try:
self.host = host
self.port = port
# Create the server, binding to self.host on port 'self.port'
self.server = socketserver.TCPServer((self.host, self.port), MyTCPHandler)
except Exception as e:
logger.error('初始化TCPServer失敗：%s' % e)
exit(1)
def start(self):
# Activate the server; this will keep running until you interrupt the program with Ctrl-C
self.server.serve_forever()
MyTCPClient.py
import socket
import configparser
import time
if __name__ == '__main__':
if_sock_connected = False
try:
config_parser = configparser.ConfigParser()
config_parser.read('./conf/tcpserver.conf', encoding='utf-8-sig')
host = config_parser.get('TCPSERVER', 'host')
port = int(config_parser.get('TCPSERVER', 'port'))
# Create a socket (SOCK_STREAM means a TCP socket)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# Connect to server and send data
sock.connect((host, port))
if_sock_connected = True # 标記socket是否已連接配接
i = 0
while i < 10000:
if i == 1000:
sock.sendall(bytes('bye\n', "utf-8"))
sock.sendall(bytes('hello world with tcp\n', "utf-8"))
# Receive data from the server
received = str(sock.recv(1024), "utf-8")
logger.info('receive data from server：%s' % received)
if received == 'bye':
time.sleep(5)
i += 1
except Exception as e:
logger.error('程式運作出錯：%s' % e)
finally:
if if_sock_connected:
sock.close()
appClient.py
#!/usr/bin/env python
from kazoo.client import KazooClient
from kazoo.client import KazooState
def my_listener(state):
if state == KazooState.LOST:
logger.info('LOST')
# Register somewhere that the session was lost
elif state == KazooState.SUSPENDED:
logger.info('SUSPENDED')
# Handle being disconnected from Zookeeper
else:
logger.info('CONNECTED')
# Handle being connected/reconnected to Zookeeper
def my_event_listener(event):
logger.info(event)
zk_client = KazooClient(hosts='10.118.52.26:2181')
zk_client.add_listener(my_listener)
zk_client.start()
node_path = '/rootNode'
sub_node = 'loaderAgent102027165'
children = zk_client.get_children(node_path, watch=my_event_listener)
logger.info('there are %s children with names %s' % (len(children), children))
@zk_client.ChildrenWatch(node_path)
def watch_children(children):
logger.info("Children are now: %s" % children)
@zk_client.DataWatch("%s/%s" % (node_path, sub_node))
def watch_node(data, state):
"""監視節點資料是否變化"""
if state:
logger.info('Version：%s, data：%s' % (state.version, data))
i = 0
while i < 1000:
time.sleep(5)
children = zk_client.get_children(node_path, watch=my_event_listener)
logger.info('there are %s children with names %s' % (len(children), children))
i += 1
zk_client.stop()
zk_client.close()
loadAgent.py
import threading
import json
import subprocess
from myTCPServer import MyTCPServer
# 全局變量
zk_conn_stat = 0 # zookeeper連接配接狀态 1-LOST 2-SUSPENDED 3-CONNECTED/RECONNECTED
registry_status = 0 # 伺服器節點在zookeeper的注冊狀态 0-未注冊、正在注冊， 1-已注冊
def restart_zk_client():
'''重新開機zookeeper會話'''
global zk_client
global zk_conn_stat
zk_client.restart()
registry_zookeeper()
logger.error('重新開機zookeeper用戶端異常：%s' % e)
def zk_conn_listener(state):
'''zookeeper連接配接狀态監聽器'''
global registry_status
logger.warn('zookeeper connection lost')
zk_conn_stat = 1
registry_status = 0 # 重置是否完成注冊
thread = threading.Thread(target=restart_zk_client)
thread.start()
logger.warn('zookeeper connection dicconnected')
zk_conn_stat = 2
zk_conn_stat = 3
logger.info('zookeeper connection cconnected/reconnected')
def registry_zookeeper():
'''注冊節點資訊到zookeeper'''
global node_parent_path
global host
global port
while zk_conn_stat != 3: # 如果zookeeper用戶端沒連上zookeeper，則先不讓注冊
continue
logger.info('正在注冊負載機到zookeeper...')
zk_client.ensure_path(node_parent_path)
loader_agent_info = '{"host":"%s", "port":%s, "status":"idle"}' % (host, port)
if not zk_client.exists('%s/loaderAgent%s' % (node_parent_path, host.replace('.', ''))):
zk_client.create('%s/loaderAgent%s' % (node_parent_path, host.replace('.', '')), loader_agent_info.encode('utf-8'), ephemeral=True, sequence=False)
# children = zk_client.get_children(node_parent_path)
# logger.info('there are %s children with names: %s' % (len(children), children))
# for child in children:
# logger.info(child)
# data, stat = zk_client.get('%s/%s' % (node_parent_path, child))
# logger.info(data)
registry_status = 1 # 完成注冊
logger.info('注冊負載機到zookeeper成功')
return True
logger.error('注冊負載機到zookeeper失敗：%s' % e)
return False
def start_tcpserver(tcpserver):
'''啟動tcp伺服器'''
tcpserver.start()
def get_server_status(proc_name):
'''通過給定程序名稱擷取伺服器狀态'''
with subprocess.Popen('ps -e | grep "%s"' % proc_name, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True, universal_newlines=True) as proc:
outs, errs = proc.communicate(timeout=30)
outs = outs.strip()
if outs.find(proc_name) != -1:
# logger.info('擷取負載機狀态成功 %s' % outs)
server_status = 'busy'
elif outs == '':
# logger.info('擷取負載機狀态成功')
server_status = 'idle'
logger.error('擷取負載機狀态失敗：%s' % errs)
server_status = 'unknow'
proc.kill()
logger.error('擷取負載機狀态失敗：%s' % e)
server_status = 'unknow'
return server_status
def update_server_status(interval, proc_name):
'''定時檢測并更新伺服器狀态：根據程序名稱是否存在來判斷伺服器狀态，如果存在則表示伺服器被占用，标記伺服器狀态為busy，否則标記伺服器狀态為 idle
如果根據程序名，檢查程序失敗，則标記伺服器狀态為unknow'''
while True:
second_for_localtime1 = time.mktime(time.localtime()) # UTC時間（秒）
if zk_conn_stat != 3: # 如果zookeeper用戶端還沒連上zookeeper，則不讓進行後續操作
if registry_status != 1: # 如果zookeeper用戶端已連上zookeeper，但是還沒注冊節點到zookeeper，則不讓進行後續操作
server_status = get_server_status(proc_name)
loader_agent_info = '{"host":"%s", "port":%s, "status":"%s"}' % (host, port, server_status)
'''
這裡為啥要加這個判斷：zookeeper删除臨時節點存在延遲，如果zookeeper用戶端主動關閉後快速重新開機并注冊節點資訊 這個過程耗時比較短，可能注冊完節點資訊時，zookeeper
還沒來得及删除重新開機之前建立的臨時節點，而本次建立的臨時節點路徑和重新開機前的一模一樣，這樣導緻的結果是，zookeeper接下來的删除操作，會把重新開機後注冊的節點也删除
'''
if zk_client.exists('%s/loaderAgent%s' % (node_parent_path, host.replace('.', ''))):
zk_client.set('%s/loaderAgent%s' % (node_parent_path, host.replace('.', '')), loader_agent_info.encode('utf-8'))
else:
registry_zookeeper()
second_for_localtime2 = time.mktime(time.localtime()) # UTC時間（秒）
time_difference = second_for_localtime2 - second_for_localtime1
if time_difference < interval:
time.sleep(interval - time_difference)
logger.info('正在啟動代理...')
logger.info('正在讀取zookeeper配置...')
config_parser.read('./conf/zookeeper.conf', encoding='utf-8-sig')
zk_hosts = config_parser.get('ZOOKEEPER', 'hosts').replace('，', ',').strip()
node_parent_path = config_parser.get('ZOOKEEPER', 'nodeParentPath').replace('，', ',').strip()
logger.info('正在建構并啟動zookeeper用戶端...')
zk_client = KazooClient(hosts=zk_hosts)
zk_client.add_listener(zk_conn_listener)
zk_client.start()
logger.error('初始化zookeeper用戶端失敗: %s' % e)
exit(1)
config_parser.clear()
tcp_server = MyTCPServer(host, port)
thread = threading.Thread(target=start_tcpserver, args=(tcp_server,))
logger.error('TCPServer啟動失敗：%s，請檢查配置/conf/tcpserver.conf是否正确' % e)
# 注冊到zookeeper
config_parser.read('./conf/agent.conf', encoding='utf-8-sig')
interval = int(config_parser.get('AGENT', 'interval'))
proc = config_parser.get('AGENT', 'proc').strip()
# 定時更新伺服器節點繁忙狀态
update_server_status(interval, proc)
logger.error('zk_client運作失敗：%s，請檢查配置/conf/agent.conf是否正确' % e)
運作效果
作者：授客
QQ：1033553122
全國軟體測試QQ交流群：7156436
Git位址：https://gitee.com/ishouke
友情提示：限于時間倉促，文中可能存在錯誤，歡迎指正、評論！
作者五行缺錢，如果覺得文章對您有幫助，請掃描下邊的二維碼打賞作者，金額随意，您的支援将是我繼續創作的源動力，打賞後如有任何疑問，請聯系我!!!
微信打賞
支付寶打賞 全國軟體測試交流QQ群