encode即编码原码按默认/gb2312/utf-8/latin-1自己写参数做选择decode()即解码让bytes流重新变成可读str这里有base64编码具体规则是data里每三个bytes24bit位对应4个6bit位流6bit位流是AL里的字符三个bytes可能在data末尾只能找到1个byte或者2个那么就补或者import base64 import json import socket import struct import threading import time from abc import ABC,abstractclassmethod message{cmd: ping, ts: 1698000000, payload: 你好 world} def my_to_bytes(data:bytes)-str: return *.join(f{b:08b} for b in data) # *.join连接 def demo_encoders(): print(the same plaintext tranverse to different bytes) # 1) JSON UTF-8文本协议人类可读HTTP/WebSocket 常用 b_jsonjson.dumps(message,ensure_asciiFalse).encode(utf-8) # Base64不是加密只是把任意 bytes 变成 64 个安全字符 # 常用于在文本协议里夹带二进制邮件附件、JWT、data URI b64base64.b64encode(b_json) print(f [JSONUTF-8 ] {len(b_json):3d} 字节 {b_json[:100]!r}...) print(f [Base64 ] {len(b64):3d} 字节 {b64[:40]!r}...) #struct 二进制打包定长、紧凑、快游戏/嵌入式/UDP 广播常用 # !H 网络字节序(大端) 无符号短整型用来做 长度前缀 body b_json framed struct.pack(!H, len(body)) body # 经典 TLV/长度前缀帧 print(f [长度前缀帧 ]{len(framed):3d} 字节 {framed[:2].hex(:,1)})#长度 (length,)struct.unpack(!H,framed[:2]) restored json.loads(framed[2:2length].decode(utf-8)) print(f [解码还原 ]{restoredmessage}-{restored}) #demo_encoders() class Transport(ABC):# 目前的理解是换名抽象代词/api_by_bike abstractclassmethod def send (self,data:bytes)-None:... def close(self)-None:pass class BadTransport(Transport): pass class Transport_Lite(): def send(self,data:bytes)-None:raise NotImplementedError def close(self)-None:pass #BadTransport #这样写直接报错 class InMemory_Transport(Transport): def __init__(self,inbox:list): self.inboxinbox def send(self,data:bytes): self.inbox.append(data) print(f [内存通道] 投递 {len(data)}字节 - {data[:24]!r}...) class TCP_Transport(Transport): def __init__(self,host127.0.0.1,port0):#只要一写TCP_Transport()__init__直接调用 self.srvsocket.socket(socket.AF_INET,socket.SOCK_STREAM) self.srv.bind((host,port)) self.srv.listen(1)#启动长期执行 self.addrself.srv.getsockname() self.receivedNone threading.Thread(targetself._serve, daemonTrue).start()#start()调用serve() # --新线程创建-- --callback 后台线程/新线程主/旧线程/程序退出后发生什么新线程仍然启动 def _serve(self):#_表示是内部 conn,_self.srv.accept()#等客户端会阻塞 hdrconn.recv(2) (n,)struct.unpack(!H,hdr) bufb while len(buf)n: bufconn.recv(n-len(buf)) self.receivedbuf conn.close() def wait_received(self,time_out1.0): ddltime.time()time_out while time.time() ddl and self.received is None: time.sleep(0.02) return self.received def send(self,data:bytes): print(f [TCP 通道] {self.addr[0]}:{self.addr[1]} send {len(data)} bytes) with socket.create_connection(self.addr) as c: c.sendall(data) def close(self): self.srv.close() def send_message(t:Transport,data:bytes)-None: print(fGoing on {type(t).__name__}) t.send(data) if isinstance(t,TCP_Transport): receivedt.wait_received() print(f [TCP 服务端] 收到并解帧 - {json.loads(received)}) t.close() if __name____main__: demo_encoders() payload json.dumps(message, ensure_asciiFalse).encode(utf-8) framed struct.pack(!H, len(payload)) payload #send_message(InMemory_Transport,framed) inbox[] send_message(InMemory_Transport(inbox),framed) send_message(TCP_Transport(),framed)这里实际涉及的协议只有frame封装但是具体的实现可以让协议变成可操作的对象离散的坑1.实现时注意__init__在调用class时直接调用2.ver1:isinstance()适合代替tInMemory_Transport()/TCP_Transport()这种写了直接调用__init__函数的写法这也是为什么写第一条的原因ver2:类不能当实例上传instance是查类的类型是否对得上不管实例化3.type(t).__name__同上4.pack 和 unpack的参数设置为!H :!代表保证整数转字节时高位字节放在整数前面;H代表装配到2个字节空间-1我不太明白socket模组的写法是如何设计的srv的具体函数不清楚thread具体函数不清楚其他可能不是新手坑。有具体理解流程的有1.pack是封装帧头unpack生成的元组第一个元素是message字节长度。2.json.loads()是对json文本转py对象3.调用ABC可以使用abstractmethod让抽象基类Transport可以在send_message()里可以不区分InMemory_Transport / TCP_Transport 实现非显示(用t:Transport)调用这是统一接口的做法。4.threading.Thread().start()是开启新线程。第一个参数写的是调用什么函数但是不写()因为这是callback写法传函数名不传函数结果第二个参数写True是子线程依附于主线程的意思总结实现这个脚本需要理解编码解码的流程和类型区分有特殊流程的base64也不过是双射规则内信息都是一串一串bytes流区别仅在于编码模式自选但是你utf-8写出的message用b64encode(message)是选错的行为;封装帧需要packjson文本转py对象需要json.loads()通信需要对线程控制这里只有callback和daemonTrue这一个例子。
