基于Socketserver+ThreadPoolExecutor+Thread構造的TCP網絡實時通信程序

介紹：

源代碼：

Socketserver-服務端代碼

Socketserver客戶端代碼：

介紹：

socketserver是一種傳統的傳輸層網絡編程接口，相比WebSocket這種應用層的協議來說，socketserver比較底層，socketserver的網絡通信邏輯與收發、傳輸的數據格式與都要由開發者自己來定義，適合用來學習網絡底層通信邏輯。我采用Python腳本來編程Socketserver的接口，我在下面放出源代碼。

源代碼：

我先講一下我實現的轉發模型，是C/S架構，不是P2P，由服務端中轉客戶端的發送消息這樣。

Socketserver-服務端代碼

import json
import socketserver
import struct
from threading import Thread
from concurrent.futures import ThreadPoolExecutorfrom threading import Lockdef send_byte(conn,msg):msg__bs_len = len(msg)msg_bs_len_bs = struct.pack('i',msg__bs_len)conn.sendall(msg_bs_len_bs)conn.sendall(msg)def recv_byte(conn):msg_recv_len_bs = conn.recv(4)msg_recv_len = struct.unpack('i', msg_recv_len_bs)[0]msg_recv = conn.recv( msg_recv_len )return msg_recvdef send(conn,msg):msg_json = json.dumps(msg)msg_bs = msg_json.encode('utf-8')msg_bs_len = len(msg_bs)msg_bs_len_pack=(struct.pack('i', msg_bs_len))conn.sendall(msg_bs_len_pack)conn.sendall(msg_bs)def recv_name(conn):name_len_bs = conn.recv(4)name_len = struct.unpack('i', name_len_bs)[0]name_bs = conn.recv(name_len)name = name_bs.decode('utf-8')return namedef recv(conn):msg_len_bs = conn.recv(4)msg_len = struct.unpack('i', msg_len_bs)[0]msg_bs = conn.recv(msg_len)msg = msg_bs.decode('utf-8')msg = json.loads(msg)return msgclass MyRequestHandler(socketserver.BaseRequestHandler):client_dict = {} #{address_port:address_port,sk_conn:conn}name_list = []stor_user_list = []retr_user_list = []lock = Lock()def handle(self):conn = self.requestaddress_port = self.client_addressclient_name = recv_name(conn)try:with ThreadPoolExecutor() as t:future = t.submit(handle_is_newuser,address_port,conn,client_name)def broadcast_welcome(future):is_new = future.result()if is_new:for key,value in MyRequestHandler.client_dict.items():sk_conn = value['sk_conn']send(sk_conn, f"系統消息: 【{client_name}】 加入了群聊，輸入/help獲取命令")future.add_done_callback(broadcast_welcome)except Exception as e:print ('出現異常:',e)while 1:msg_dict = recv(conn)print (msg_dict)msg = msg_dict['msg']name = msg_dict['name']try:if msg.upper() == 'Q':MyRequestHandler.name_list.remove(client_name)del MyRequestHandler.client_dict[name]for key, value in MyRequestHandler.client_dict.items():sk_conn = value['sk_conn']print (f'【{name}】退出了群聊')send(sk_conn, f'【{name}】退出了群聊')conn.close()elif msg == 'client/all':send(conn,f'在線用戶列表:{MyRequestHandler.name_list}')elif msg == '/help':text ='查看在線用戶:client/all\n私聊:/chat [對方名字] [消息內容]\n退出群聊:[q] or [Q]\n向對方傳輸文件:/stor [對方名字] [本地文件路徑]\n顯示遞歸目錄樹:/tree [對方名字] [遠端目錄]'send(conn,text)elif msg.lstrip().startswith('/tree_content'):try:parts = msg.split(' ',2)ip_or_name = parts[1]if ip_or_name == name:send(conn,'請指定對方名字')continueif ip_or_name in MyRequestHandler.name_list:values = MyRequestHandler.client_dict[ip_or_name]pri_conn = values['sk_conn']send(pri_conn,msg_dict)except Exception as e:print ('命令執行錯誤',e)elif msg.lstrip().startswith('/tree'):parts = msg.split(' ',2)ip_or_name = parts[1]if ip_or_name == name:send(conn, '請指定對方名字')continueif ip_or_name in MyRequestHandler.name_list:values = MyRequestHandler.client_dict[ip_or_name]remote_conn = values['sk_conn']send(remote_conn,msg_dict)continueelif msg.lstrip().startswith('stor')  or  msg.lstrip().startswith('retr') :print ('第一次文件傳輸交互')msg_bytes = recv_byte(conn)parts = msg.split(' ',3)remote_name= parts[1]client_dict_value = MyRequestHandler.client_dict[remote_name]remote_conn = client_dict_value['sk_conn']cmd = parts[0]send(remote_conn,msg_dict)if cmd == '/stor':print('進來了')send_byte(remote_conn,msg_bytes)print (msg_bytes)print ('發送成功')continueelse:for key, value in MyRequestHandler.client_dict.items():sk_conn = value['sk_conn']send(sk_conn, msg_dict)except Exception as e:print ('意外報錯:',e)def handle_is_newuser(address_port,conn,client_name):dict_addr_conn = {}with MyRequestHandler.lock:if client_name in MyRequestHandler.name_list:returnelse:dict_addr_conn['address_port'] = address_portdict_addr_conn['sk_conn'] = connMyRequestHandler.client_dict[client_name] = dict_addr_connMyRequestHandler.name_list.append(client_name)return Trueif __name__ == '__main__':server = socketserver.ThreadingTCPServer(('127.0.0.1', 12345), MyRequestHandler)print("服務器正在運行...")server.serve_forever()

Socketserver客戶端代碼：
?

import json
import os
import socket
import struct
from threading import Thread
import sys
import  timename = ''
stor_user_list=[]def send_byte(conn,msg):msg__bs_len = len(msg)msg_bs_len_bs = struct.pack('i',msg__bs_len)conn.sendall(msg_bs_len_bs)conn.sendall(msg)def recv_byte(conn):msg_recv_len_bs = conn.recv(4)msg_recv_len = struct.unpack('i', msg_recv_len_bs)[0]msg_recv = conn.recv( msg_recv_len)return msg_recvdef send_name(conn):global namename = input('請取個名字吧:')name_bs = name.encode('utf-8')name_len = len(name_bs)conn.sendall(struct.pack('i', name_len))conn.sendall(name_bs)def send_handle(conn,name_msg):name_msg_json = json.dumps(name_msg)name_msg_json_bs = name_msg_json.encode('utf-8')name_msg_json_bs_len = len(name_msg_json_bs)name_msg_json_bs_len_pack = struct.pack('i', name_msg_json_bs_len)conn.sendall(name_msg_json_bs_len_pack)conn.sendall(name_msg_json_bs)def send(conn):global stor_user_listwhile True:name_msg = {}msg = input()name_msg['name'] = namename_msg['msg'] = msgtry:if msg.upper() == 'Q':# name_msg_json = json.dumps(name_msg)# msg_bs = name_msg_json.encode('utf-8')# msg_len = len(msg_bs)# conn.sendall(struct.pack('i', msg_len))# conn.sendall(msg_bs)send_handle(conn,name_msg)print ('我退出了群聊！')conn.close()sys.exit()if  str(msg.lstrip()).startswith('/stor') or  str(msg.lstrip()).startswith('/retr') :print('主動發起文件傳輸（A端）')parts = msg.split(' ', 3)command = parts[0]remote_name = parts[1]localpath = parts[2]# name_msg_json = json.dumps(name_msg)# msg_json_bs = name_msg_json.encode('utf-8')## msg_json_bs_len = len(msg_json_bs)# msg_json_bs_len_pack = struct.pack('i', msg_json_bs_len)## conn.sendall(msg_json_bs_len_pack)# conn.sendall(msg_json_bs )if '/stor' in command:name_byte = {}name_byte['name'] = namename_byte['msg'] = msgsend_handle(conn,name_byte)with open(localpath, mode='rb') as read_file:bytes = read_file.read()print('開始發送文件')send_byte(conn,bytes)print('文件發送成功')sys.stdout.write(f'{name}>>')sys.stdout.flush()continuesend_handle(conn,name_msg)sys.stdout.write(f'{name}>>')sys.stdout.flush()except Exception as e:print('異常報錯：', e)sys.exit()def recv_handle(conn):msg_len_pack = conn.recv(4)msg_bs_len = struct.unpack('i', msg_len_pack)[0]msg_bs = conn.recv(msg_bs_len)msg_dict_json = msg_bs.decode('utf-8')msg_dict = json.loads(msg_dict_json)return msg_dictdef recv(conn):global stor_user_listwhile True:try:msg_dict = recv_handle(conn)sys.stdout.write('\r' + ' ' * 100 + '\r')  # 覆蓋當前行sys.stdout.flush()if isinstance(msg_dict,list):print (msg_dict)elif isinstance(msg_dict,str):#由服務器發送的消息，因此無需以字典格式傳輸print (msg_dict)elif isinstance(msg_dict,dict):msg = msg_dict['msg']name_msg = msg_dict['name']print (name_msg)if msg.lstrip().startswith('/chat'):parts = msg.split(' ', 2)pri_msg = parts[2]print(f'{name_msg}>>{name} {pri_msg}')if msg.lstrip().startswith('/tree'):parts = msg.split(' ', 2)local_path = parts[2]tree =''def list_tree(path,tree,depth=1):dir_name = os.path.basename(path)tree += str(depth * '|-----')+str(dir_name).strip() + '\n'file_list = os.listdir(path)for file in file_list:filepath = os.path.join(path,file)if os.path.isdir(filepath):tree = list_tree(filepath,tree,depth+1)if os.path.isfile(filepath):tree += str(depth * '|-----') + '|-----' + file + '\n'return treedir_tree =list_tree(local_path,tree)dir_tree_full = '\n' + dir_treeprint (dir_tree_full)msg_dir_tree = {}msg_dir_tree['name'] = namemsg_dir_tree['msg'] = dir_tree_fullsend_handle(conn,msg_dir_tree)if name_msg != name and msg.upper() != 'Q' and  not msg.lstrip().startswith('/chat') and not  msg.lstrip().startswith('stor')  and not msg.lstrip().startswith('retr'):print(f'{name_msg}>> {msg}')if msg.lstrip().startswith('stor') or msg.lstrip().startswith('retr'):msg_bytes = recv_byte(conn)parts = msg_dict['msg'].split(' ',3)command = parts[0]local_path = parts[3]if  '/stor' in command:with open(local_path, mode='wb') as writefile:print('開始文件傳輸（B端）',flush=True)writefile.write(msg_bytes)writefile.flush()os.fsync(writefile.fileno())print ('傳輸完畢', flush=True)sys.stdout.write(f'{name}>>')sys.stdout.flush()except Exception as e:print ('接收消息出錯:',e)if __name__ == '__main__':try:sk = socket.socket()sk.connect(('127.0.0.1', 12345))except Exception as e:print ('socket連接失敗',e)sys.exit()send_name(sk)receiver = Thread(target=recv, args=(sk,), daemon=True)receiver.start()send(sk)sk.close()