借助zookeeper可以實(shí)現(xiàn)服務(wù)器的注冊(cè)與發(fā)現(xiàn),有需求的時(shí)候調(diào)用zookeeper來(lái)發(fā)現(xiàn)可用的服務(wù)器,將任務(wù)均勻分配到各個(gè)服務(wù)器上去.

這樣可以方便的隨任務(wù)的繁重程度對(duì)服務(wù)器進(jìn)行彈性擴(kuò)容,客戶(hù)端和服務(wù)端是非耦合的,也可以隨時(shí)增加客戶(hù)端.

zk_server.py

import threadingimport jsonimport socketimport sysfrom kazoo.client import KazooClient# TCP服務(wù)端綁定端口開(kāi)啟監(jiān)聽(tīng)，同時(shí)將自己注冊(cè)到zkclass 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): ''' 開(kāi)始服務(wù)，每次獲取得到一個(gè)信息，都新建一個(gè)線程處理 ''' self.sock.listen(128) self.register_zk() print('開(kāi)始監(jiān)聽(tīng)') while True: conn, addr = self.sock.accept() print('建立鏈接%s' % str(addr)) t = threading.Thread(target=self.handle, args=(conn, addr)) t.start() # 具體的處理邏輯,只要接收到數(shù)據(jù)就立即投入工作，下次沒(méi)有數(shù)據(jù)本次鏈接結(jié)束 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!!!’) # 將自己注冊(cè)到zk，臨時(shí)節(jié)點(diǎn)，所以連接不能中斷 def register_zk(self): ''' 注冊(cè)到zookeeper ''' self.zk = KazooClient(hosts=’127.0.0.1:2181’) self.zk.start() self.zk.ensure_path(’/rpc’) # 創(chuàng)建根節(jié)點(diǎn) value = json.dumps({’host’: self.host, ’port’: self.port}) # 創(chuàng)建服務(wù)子節(jié)點(diǎn) 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 randomimport sysimport timeimport jsonimport socketfrom kazoo.client import KazooClient# 客戶(hù)端連接zk,并從zk獲取可用的服務(wù)器列表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獲取服務(wù)器地址信息列表 ''' 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): ''' 隨機(jī)選出一個(gè)可用的服務(wù)器 ''' return random.choice(self._servers) def get_connection(self): ''' 提供一個(gè)可用的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 sockif __name__ == ’__main__’: # 模擬多個(gè)客戶(hù)端批量生成任務(wù)，推送給服務(wù)器執(zhí)行 client = ZKClient() for i in range(40): sock = client.get_connection() sock.send(bytes(str(i), encoding=’utf8’)) sock.close() time.sleep(1)

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持好吧啦網(wǎng)。