溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
同樣在services.py 中自定義需要實現消息協議、傳輸控制，并且實現客戶端存根clientStub和服務器端存根serverStub，服務器定義以及channel的定義。
此時，添加DistributedChannel分布式的channel，并在ThreadServer中添加了registry方法。
```
import threading
import random
import struct
import time
from io import BytesIO
import socket
import json
from kazoo.client import KazooClient
class InvalidOperation(BaseException):
def __init__(self, message = None):
self.message = message or 'involid operation'
class MethodProtocol(object):
''''
解讀方法名
'''
def __init__(self, connection):
self.conn = connection
def _read_all(self, size):
"""
幫助我們讀取二進制數據
:param size: 想要讀取的二進制數據大小
:return: 二進制數據bytes
"""
# self.conn
if isinstance(self.conn, BytesIO):
buff = self.conn.read(size)
return buff
else:
# 有時候長度大于每次讀取的長度
have = 0
buff = b''
while have < size:
chunk = self.conn.recv(size - have)
buff += chunk
l = len(chunk)
have += l
if l == 0:
# 表示客戶端已經關閉了
raise EOFError
return buff
def get_method_name(self):
# 讀取字符串長度
buff = self._read_all(4)
length = struct.unpack('!I',buff)[0]
# 讀取字符串
buff = self._read_all(length)
name = buff.decode()
return name
class DivideProtocol(object):
"""
divide過程消息協議轉換工具
"""
def args_encode(self, num1, num2=1):
"""
將原始調用的請求參數轉換打包成二進制消息數據
:param num1: int
:param num2: int
:return: bytes 二進制消息數據
"""
name = 'divide'
# 處理函數名
buff = struct.pack('!I', 6) # 無符號int
buff += name.encode()
# 處理參數1
buff2 = struct.pack('!B', 1) # 無符號byte
buff2 += struct.pack('!i', num1)
# 處理參數2
if num2 != 1:
# 沒有傳參的時候
buff2 += struct.pack('!B', 2)
buff2 += struct.pack('!i', num2)
# 處理參數邊界和組合成完整數據
buff += struct.pack('!I',len(buff2))
buff += buff2
return buff
def _read_all(self, size):
"""
幫助我們讀取二進制數據
:param size: 想要讀取的二進制數據大小
:return: 二進制數據bytes
"""
# self.conn
if isinstance(self.conn, BytesIO):
buff = self.conn.read(size)
return buff
else:
# 有時候長度大于每次讀取的長度
have = 0
buff = b''
while have < size:
chunk = self.conn.recv(size - have)
buff += chunk
l = len(chunk)
have += l
if l == 0:
# 表示客戶端已經關閉了
raise EOFError
return buff
def args_decode(self, connection):
"""
接受調用請求數據病進行解析
:param connection: 鏈接請求數據 socket BytesIO
:return: 因為有多個參數,定義為字典
"""
param_len_map = {
1:4,
2:4,
}
param_fmt_map = {
1:'!i',
2:'!i',
}
param_name_map = {
1: 'num1',
2: 'num2',
}
# 保存用來返回的參數字典
args = {}
self.conn = connection
# 處理方法的名字,已經提前被處理，稍后處理
# 處理消息邊界
# 1) 讀取二進制數據----read ， ------ByteIO.read
# 2) 將二進制數據轉換為python的數據類型
buff = self._read_all(4)
length = struct.unpack('!I',buff)[0]
# 記錄已經讀取的長度值
have = 0
# 處理第一個參數
# 解析參數序號
buff = self._read_all(1)
have += 1
param_seq = struct.unpack('!B', buff)[0]
# 解析參數值
param_len = param_len_map[param_seq]
buff = self._read_all(param_len)
have += param_len
param_fmt = param_fmt_map[param_seq]
param = struct.unpack(param_fmt,buff)[0]
# 設置解析后的字典
param_name = param_name_map[param_seq]
args[param_name] = param
if have >= length:
return args
# 處理第二個參數
# 解析參數序號
buff = self._read_all(1)
param_seq = struct.unpack('!B', buff)[0]
# 解析參數值
param_len = param_len_map[param_seq]
buff = self._read_all(param_len)
param_fmt = param_fmt_map[param_seq]
param = struct.unpack(param_fmt, buff)[0]
# 設置解析后的字典
param_name = param_name_map[param_seq]
args[param_name] = param
return args
def result_encode(self, result):
"""
將原始結果數據轉換為消息協議二進制數據
:param result:
:return:
"""
if isinstance(result,float):
# 處理返回值類型
buff = struct.pack('!B', 1)
buff += struct.pack('!f', result)
return buff
else:
buff = struct.pack('!B', 2)
# 處理返回值
length = len(result.message)
# 處理字符串長度
buff += struct.pack('!I', length)
buff += result.message.encode()
return buff
def result_decode(self, connection):
"""
將返回值消息數據轉換為原始返回值
:param connection: socket BytesIo
:return: float InvalidOperation對象
"""
self.conn = connection
# 處理返回值類型
buff = self._read_all(1)
result_type = struct.unpack('!B', buff)[0]
if result_type == 1:
#正常情況
buff = self._read_all(4)
val = struct.unpack('!f', buff)[0]
return val
else:
buff = self._read_all(4)
length = struct.unpack('!I', buff)[0]
# 讀取字符串
buff = self._read_all(length)
message = buff.decode()
return InvalidOperation(message)
class Channel(object):
"""
用于客戶端建立網絡鏈接
"""
def __init__(self, host, port):
self.host = host
self.port = port
def get_connection(self):
"""
獲取鏈接對象
:return: 與服務器通訊的socket
"""
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect((self.host, self.port))
return sock
class DistributedChannel(object):
"""
支持分布式的zookeeper的RPC客戶端鏈接工具
"""
def __init__(self):
# 創建kazoo對象，用來跟zookeeper鏈接，獲取信息
zk = KazooClient('127.0.0.1:2181')
zk.start()
self.zk = zk
self._servers = []
self._get_servers() # 第一次，手動開啟
def _get_servers(self, event=None):
"""
從zookeeper中獲取所有可用的RPC服務器的地址
:return:
"""
self._servers = [] # 每次重新獲取地址信息
# 從zookeeper中獲取/rpc節點下的所有可用的rpc服務器節點
servers = self.zk.get_children('/rpc', watch=self._get_servers) # 監視的回調函數為自身
for server in servers:
addr_data = self.zk.get('/rpc/' + server)[0]
addr = json.loads(addr_data.decode())
self._servers.append(addr)
def _get_server(self):
"""
從可用的服務器列表中選出一臺服務器
:return: {"host":xxx,"port":xxx}
"""
return random.choice(self._servers)
def get_connection(self):
"""
提供一個具體的與RPC服務器的鏈接socket
:return:
"""
while True:
addr = self._get_server()
print(addr)
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect((addr['host'], addr['port']))
except ConnectionRefusedError:
time.sleep(1)
continue
else:
return sock
class ThreadServer(object):
"""
多線成RPC服務器
"""
def __init__(self, host, port, handlers):
sock = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
# 地址復用
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.host = host
self.port = port
# 綁定地址
sock.bind((self.host, self.port))
# 因為在啟動的方法中才開啟監聽，所以不在此處開啟
# sock.listen(128)
self.sock = sock
self.handlers = handlers
def serve(self):
"""
開啟服務器運行，提供RPC服務
:return:
"""
# 開啟服務器的監聽，等待客戶端的鏈接請求
self.sock.listen(128)
print("服務器開啟監聽,ip地址為%s,port為%d..." % (self.host,self.port))
# 注冊到zookeeper
self.register_zookeeper()
while True:
# 不斷的接收客戶端的鏈接請求
client_sock, client_addr = self.sock.accept()
print("與客戶端%s建立連接" % str(client_addr))
t = threading.Thread(target= self.handle, args=(client_sock,))
t.start()
def register_zookeeper(self):
"""
在zookeeper中心注冊本服務器的地址信息
:return:
"""
# 創建kazoo的客戶端
zk = KazooClient('127.0.0.1:2181')
# 建立與zookeeper的鏈接
zk.start()
# 在zookeeper中創建節點保存數據
zk.ensure_path('/rpc')
data = json.dumps({'host':self.host,'port':self.port})
zk.create('/rpc/server', data.encode(), ephemeral=True, sequence=True)
# 子線程函數
def handle(self,client_sock):
"""
子線程調用的方法，用來處理一個客戶段的請求
:return:
"""
# 交個ServerStub,完成客戶端的具體的RPC的調用請求
stub = ServerStub(client_sock, self.handlers)
try:
while True:
# 不斷的接收
stub.process()
except EOFError:
# 表示客戶端關閉了連接
print('客戶端關閉了連接')
client_sock.close()
class Server(object):
"""
RPC服務器
"""
def __init__(self, host, port, handlers):
sock = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
# 地址復用
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.host = host
self.port = port
# 綁定地址
sock.bind((self.host, self.port))
# 因為在啟動的方法中才開啟監聽，所以不在此處開啟
# sock.listen(128)
self.sock = sock
self.handlers = handlers
def serve(self):
"""
開啟服務器運行，提供RPC服務
:return:
"""
# 開啟服務器的監聽，等待客戶端的鏈接請求
self.sock.listen(128)
print("服務器開啟監聽,ip地址為%s,port為%d..." % (self.host,self.port))
while True:
# 不斷的接收客戶端的鏈接請求
client_sock, client_addr = self.sock.accept()
print("與客戶端%s建立連接" % str(client_addr))
# 交個ServerStub,完成客戶端的具體的RPC的調用請求
stub = ServerStub(client_sock, self.handlers)
try:
while True:
# 不斷的接收
stub.process()
except EOFError:
# 表示客戶端關閉了連接
print('客戶端關閉了連接')
client_sock.close()
class ClientStub(object):
"""
用來幫助客戶端完成遠程過程調用 RPC調用
stub = ClientStub()
stub.divide(200, 100)
"""
def __init__(self, channel):
self.channel = channel
self.conn = self.channel.get_connection()
def divide(self, num1, num2 = 1):
# 將調用的參數打包成消息協議的數據
proto = DivideProtocol()
args = proto.args_encode(num1, num2)
# 將消息數據通過網絡發送給服務器
self.conn.sendall(args)
# 接受服務器返回的消息數據，并進行解析
result = proto.result_decode(self.conn)
# 將結果之（正常float 或 異常InvalidOperation）返回給客戶端
if isinstance(result,float):
return result
else:
raise result
class ServerStub(object):
"""
服務端存根
幫助服務端完成遠端過程調用
"""
def __init__(self, connection, handlers):
"""
:param connection: 與客戶端的鏈接
:param handlers: 真正的本地函數路由
此處不以map的形式處理，實現類的形式
class Handler：
@staticmethod
def divide():
pass
@staticmethod
def add():
pass
"""
self.conn = connection
self.method_proto = MethodProtocol(self.conn)
self.process_map = {
'divide': self._process_divide,
'add': self._process_add
}
self.handlers = handlers
def process(self):
"""
當服務端接受了客戶的鏈接，建立好鏈接后，完成遠端調用的處理
:return:
"""
# 接收消息數據，并解析方法的名字
name = self.method_proto.get_method_name()
# 根據解析獲得的方法名，調用相應的過程協議，接收并解析消息數據
self.process_map[name]()
def _process_divide(self):
"""
處理除法過程調用
:return:
"""
proto = DivideProtocol()
args = proto.args_decode(self.conn)
# args = {'num1':xxx, 'num2':xxx}
# 除法過程的本地調用------------------->>>>>>>>>
# 將本地調用過程的返回值(包括可能的異常)打包成消息協議的數據,通過網絡返回給客戶端
try:
val = self.handlers.divide(**args)
except InvalidOperation as e:
ret_message = proto.result_encode(e)
else:
ret_message = proto.result_encode(val)
self.conn.sendall(ret_message)
def _process_add(self):
"""
處理加法過程調用
此方法暫時不識閑
:return:
"""
pass
if __name__ == '__main__':
# 目的：消息協議測試，模擬網絡傳輸
# 構造消息數據
proto = DivideProtocol()
# 測試一
# divide(200,100)
# message = proto.args_encode(200,100)
# 測試二
message = proto.args_encode(200)
conn = BytesIO()
conn.write(message)
conn.seek(0)
# 解析消息數據
method_proto = MethodProtocal(conn)
name = method_proto.get_method_name()
print(name)
args = proto.args_decode(conn)
print(args)```
接下來，修改server.py文件
server.py
```
from services import InvalidOperation
# from services import Server
from services import ThreadServer
import sys
class Handlers:
@staticmethod
def divide(num1, num2 = 1):
if num2 == 0:
raise InvalidOperation('ck_god_err')
val = num1/num2
return val
if __name__ == '__main__':
# 開啟服務器
# _server = Server('127.0.0.1', 8000, Handlers)
# _server.serve()
# 從啟動命令中提取服務器運行的ip地址和端口號,啟動的多線程服務器
host = sys.argv[1]
port = int(sys.argv[2])
_server = ThreadServer(host, port, Handlers)
_server.serve()```
最后，將 client.py文件也稍作修改。
```
import time
from services import ClientStub
# from services import Channel
from services import DistributedChannel
from services import InvalidOperation
# 創建與服務器的連接
# channel = Channel('127.0.0.1', 8000)
channel = DistributedChannel()
# 進行調用
for i in range(50):
try:
# 創建用于rpc調用的工具
stub = ClientStub(channel) # 初始化的時候才真正的創建連接了，所以放到里面
val = stub.divide(i * 100,100)
# val = stub.divide(i * 100)
# val = stub.divide( 100, 0)
except InvalidOperation as e:
print(e.message)
else:
print(val)
time.sleep(1)```
ctrl + shift + T在pycharm中打開多個Terminal窗口
右鍵運行客戶端，可以看到不斷地隨機切換服務器。
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
