appClient.py
#!/usr/bin/env python
#-*- encoding:utf-8 -*-
__author__ = 'shouke'
import time
from log import logger
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
#!/usr/bin/env python 3.4.0
#-*- encoding:utf-8 -*-
__author__ = 'shouke'
import time
import threading
import configparser
import json
import subprocess
from kazoo.client import KazooClient
from kazoo.client import KazooState
from log import logger
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
try:
zk_client.restart()
registry_zookeeper()
except Exception as e:
logger.error('重启zookeeper客户端异常：%s' % e)
def zk_conn_listener(state):
'''zookeeper连接状态监听器'''
global zk_conn_stat
global registry_status
if state == KazooState.LOST:
logger.warn('zookeeper connection lost')
zk_conn_stat = 1
registry_status = 0 # 重置是否完成注册
# Register somewhere that the session was lost
thread = threading.Thread(target=restart_zk_client)
thread.start()
elif state == KazooState.SUSPENDED:
logger.warn('zookeeper connection dicconnected')
zk_conn_stat = 2
# Handle being disconnected from Zookeeper
else:
zk_conn_stat = 3
logger.info('zookeeper connection cconnected/reconnected')
# Handle being connected/reconnected to Zookeeper
def registry_zookeeper():
'''注册节点信息到zookeeper'''
global node_parent_path
global host
global port
global zk_client
global zk_conn_stat
global registry_status
try:
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
except Exception as e:
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:
try:
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'
else:
logger.error('获取负载机状态失败：%s' % errs)
server_status = 'unknow'
except Exception as e:
proc.kill()
logger.error('获取负载机状态失败：%s' % e)
server_status = 'unknow'
return server_status
def update_server_status(interval, proc_name):
'''定时检测并更新服务器状态：根据进程名称是否存在来判断服务器状态，如果存在则表示服务器被占用，标记服务器状态为busy，否则标记服务器状态为 idle
如果根据进程名，检查进程失败，则标记服务器状态为unknow'''
global node_parent_path
global host
global port
while True:
second_for_localtime1 = time.mktime(time.localtime()) # UTC时间（秒）
if zk_conn_stat != 3: # 如果zookeeper客户端还没连上zookeeper，则不让进行后续操作
continue
if registry_status != 1: # 如果zookeeper客户端已连上zookeeper，但是还没注册节点到zookeeper，则不让进行后续操作
continue
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)
if __name__ == '__main__':
logger.info('正在启动代理...')
try:
logger.info('正在读取zookeeper配置...')
config_parser = configparser.ConfigParser()
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()
except Exception as e:
logger.error('初始化zookeeper客户端失败: %s' % e)
exit(1)
try:
config_parser.clear()
config_parser.read('./conf/tcpserver.conf', encoding='utf-8-sig')
host = config_parser.get('TCPSERVER', 'host')
port = int(config_parser.get('TCPSERVER', 'port'))
tcp_server = MyTCPServer(host, port)
thread = threading.Thread(target=start_tcpserver, args=(tcp_server,))
thread.start()
except Exception as e:
logger.error('TCPServer启动失败：%s，请检查配置/conf/tcpserver.conf是否正确' % e)
exit(1)
try:
# 注册到zookeeper
registry_zookeeper()
config_parser.clear()
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)
except Exception as e:
logger.error('zk_client运行失败：%s，请检查配置/conf/agent.conf是否正确' % e)
exit(1)
运行效果
