Python版本Zinx——（5）消息封装（封包、拆包）
2023-08-27 本文已影响0人
爱折腾的胖子
最近想研究一下关于长链接的相关内容，在B站上看到了Zinx框架的视频，是Golang语言的框架，本着更好的理解框架的内容，按照整个Zinx课程的进度，制作一个Python版本的Zinx框架。有关于Zinx框架的具体内容，可以看框架作者的介绍。
python版本的Zinx，基于Gevent 22.10.2，使用协程功能。
golang版本的Zinx项目，项目中两个文件夹，ziface和znet。
- ziface主要是存放一些Zinx框架的全部模块的抽象层接口类。
- znet模块是zinx框架中网络相关功能的实现，所有网络相关模块都会定义在znet模块中。
└── zinx
├── ziface
│ └──
└── znet
├──
python中的关键字没有interface，但是可以使用抽象基类(abstract base class)和第三方库来实现类似于接口的功能。在实际开发中，我们可以根据具体需求选择合适的实现方式。
暂时使用抽象基类的形式模拟接口的实现。
上一节封装了全局的配置，这节做一下消息的封装。在封装request的时候关于数据并没有做过多的处理。这节继续丰富一下，将数据的分类、长度、数据进行打包处理。
在ziface中创建imessage.py文件。
# -*- coding: utf-8 -*-
from abc import ABC, abstractmethod
class IMessage(ABC):
"""
将请求的一个消息封装到message中，定义抽象层接口
"""
@abstractmethod
def GetDataLen(self) -> int:
"""
获取消息数据段长度
:return:
"""
pass
@abstractmethod
def GetMsgId(self) -> int:
"""
获取消息ID
:return:
"""
pass
@abstractmethod
def GetData(self) -> bytes:
"""
获取消息内容
:return:
"""
pass
@abstractmethod
def SetDataLen(self, m_len: int):
"""
设置消息数据段长度
:param m_len:
"""
pass
@abstractmethod
def SetMsgId(self, m_id: int):
"""
设置消息ID
:param m_id:
"""
pass
@abstractmethod
def SetData(self, data: bytes):
"""
设置消息内容
:param data:
"""
pass
在znet中做一个实现类。
# -*- coding: utf-8 -*-
from ziface.imessage import IMessage
class Message(IMessage):
"""
将请求的一个消息封装到message中，定义抽象层接口
"""
def __init__(self, m_id: int, m_len: int, m_data: bytes):
self.id: int = m_id
self.length: int = m_len
self.data: bytes = m_data
def GetDataLen(self) -> int:
"""
获取消息数据段长度
:return:
"""
return self.length
def GetMsgId(self) -> int:
"""
获取消息ID
:return:
"""
return self.id
def GetData(self) -> bytes:
"""
获取消息内容
:return:
"""
return self.data
def SetDataLen(self, m_len: int):
"""
设置消息数据段长度
:param m_len:
"""
self.length = m_len
def SetMsgId(self, m_id: int):
"""
设置消息ID
:param m_id:
"""
self.id = m_id
def SetData(self, m_data: bytes):
"""
设置消息内容
:param m_data:
"""
self.data = m_data
def NewMessage(m_id: int, m_len: int, m_data: bytes) -> IMessage:
ns = Message(m_id, m_len, m_data)
return ns
接下来做封包、拆包。统一数据格式：数据长度（4位）|数据ID（4位）|数据内容
在ziface中创建idatapack。
# -*- coding: utf-8 -*-
from ziface.imessage import IMessage
from abc import ABC, abstractmethod
class IDataPack(ABC):
"""
封包数据和拆包数据
直接面向TCP连接中的数据流,为传输数据添加头部信息，用于处理TCP粘包问题。
"""
@abstractmethod
def GetHeadLen(self) -> int:
"""
获取包头长度方法
:return:
"""
pass
@abstractmethod
def Pack(self, msg: IMessage) -> bytes:
"""
封包方法
:param msg:
:return:
"""
pass
@abstractmethod
def Unpack(self, data: bytes) -> IMessage:
"""
拆包方法
:param data:
:return:
"""
pass
在znet中，做实现类。
# -*- coding: utf-8 -*-
from typing import Optional
from ziface.imessage import IMessage
from znet.message import NewMessage
from ziface.idatapack import IDataPack
from utils.globalobj import GlobalObject
class DataPack(IDataPack):
"""
采用经典的TLV(Type-Len-Value)封包格式
每一个数据包的格式如下：
DataLen | ID | Data
数据长度 | 数据ID | 数据内容
每一个数据包的格式统一，从TCP流中读取数据，就按照这个格式进行解析。
"""
def __init__(self):
pass
def GetHeadLen(self) -> int:
"""
获取包头长度方法
:return:
"""
# Id uint32(4字节) + DataLen uint32(4字节)
return 8
def Pack(self, msg: IMessage) -> bytes:
"""
封包方法
:param msg:
:return:
"""
# 创建一个存放bytes字节的缓冲
buffer = bytes()
# 写dataLen
buffer += msg.GetDataLen().to_bytes(4, "little")
# 写msgID
buffer += msg.GetMsgId().to_bytes(4, "little")
# 写data数据
buffer += msg.GetData()
return buffer
def Unpack(self, data: bytes) -> Optional[IMessage]:
"""
拆包方法
:param data:
:return:
"""
# 读dataLen
data_len = int.from_bytes(data[:4], byteorder='little', signed=False)
# 度msgID
msg_id = int.from_bytes(data[4:], byteorder='little', signed=False)
# 判断dataLen的长度是否超出我们允许的最大包长度
if 0 < GlobalObject.MaxPacketSize < data_len:
return None
msg = NewMessage(msg_id, data_len, bytes())
# 这里只需要把head的数据拆包出来就可以了，然后再通过head的长度，再从conn读取一次数据
return msg
def NewDataPack() -> IDataPack:
dp = DataPack()
return dp
封装好的消息处理，在IRequest中集成一下。
# -*- coding: utf-8 -*-
from ziface.iconnection import IConnection
from abc import ABC, abstractmethod
class IRequest(ABC):
"""
IRequest 接口：实际上是把客户端请求的链接信息 和 请求的数据 包装到了 Request里
"""
@abstractmethod
def GetConnection(self) -> IConnection:
"""
获取请求连接信息
:return:
"""
pass
@abstractmethod
def GetData(self) -> bytes:
"""
获取请求消息的数据
:return:
"""
pass
@abstractmethod
def GetMsgID(self) -> int:
"""
获取请求消息的ID
:return:
"""
pass
封装好的消息处理，在Request中集成一下。
# -*- coding: utf-8 -*-
from ziface.irequest import IRequest
from ziface.imessage import IMessage
from ziface.iconnection import IConnection
class Request(IRequest):
def __init__(self, conn: IConnection, msg: IMessage):
"""
:param conn: 已经和客户端建立好的 链接
:param msg: 客户端请求的数据
"""
self.Conn: IConnection = conn
self.Msg: IMessage = msg
def GetConnection(self) -> IConnection:
"""
获取请求连接信息
:return:
"""
return self.Conn
def GetData(self) -> bytes:
"""
获取请求消息的数据
:return:
"""
return self.Msg.GetData()
def GetMsgID(self) -> int:
"""
获取请求消息的ID
:return:
"""
return self.Msg.GetMsgId()
def NewRequest(conn: IConnection, msg: IMessage) -> IRequest:
req = Request(conn, msg)
return req
Request修改了，调用Request的位置需要相应的修改。修改Connection的StartReader方法。
def StartReader(self):
"""
处理读业务
:return:
"""
print("开启读业务")
while True:
try:
dp = NewDataPack()
# 读取客户端的Msg head
head_data = self.Conn.recv(dp.GetHeadLen())
if len(head_data) == 0:
# head_data 长度为0表示客户端已经退出
break
# 拆包，得到msgID 和 dataLen 放在msg中
msg = dp.Unpack(head_data)
# 根据 dataLen 读取 data，放在msg.Data中
data_content = self.Conn.recv(msg.GetDataLen())
msg.SetData(data_content)
# 得到当前conn数据的Request请求数据
req = NewRequest(self, msg)
g2 = gevent.spawn(self.RunHandler(req))
GlobalGevents.append(g2)
except Exception as e:
print("读取数据异常 ", e)
break
print("读业务关闭，ID", self.ConnID)
self.Stop()
拆包封装好了，接下来再IConnection提供一个发送封包方法SendMsg。
# -*- coding: utf-8 -*-
import socket
from abc import ABC, abstractmethod
class IConnection(ABC):
@abstractmethod
def Start(self):
"""
启动链接 让当前的链接准备开始工作
:return:
"""
pass
@abstractmethod
def Stop(self):
"""
停止链接 结束当前链接的工作
:return:
"""
pass
@abstractmethod
def GetConnID(self) -> int:
"""
获取当前链接模块的链接ID
:return:
"""
pass
@abstractmethod
def GetTCPConnection(self) -> socket.socket:
"""
获取当前链接绑定的socket conn
:return:
"""
pass
@abstractmethod
def GetRemoteAddr(self) -> tuple:
"""
获取远程客户端的TCP状态 IP Port
:return:
"""
pass
@abstractmethod
def SendMsg(self, msgID: int, data: bytes):
"""
发送数据 将数据发送给远程的客户端
:param msgID:
:param data:
:return:
"""
pass
在Connection中实现一下。
def SendMsg(self, msgID: int, data: bytes):
"""
发送数据 将数据发送给远程的客户端
:param msgID:
:param data:
:return:
"""
if self.is_closed:
raise Exception("发送数据时客户端链接被关闭")
try:
dp = NewDataPack()
msg = dp.Pack(NewMessage(msgID, len(data), data))
self.Conn.send(msg)
except Exception as e:
print(e)
服务接口做完了。在demo\datapack中做一个客户端client.py测试一下
import socket
import time
import sys
sys.path.append("../../")
from znet.datapack import NewDataPack
from znet.message import NewMessage
if __name__ == '__main__':
HOST = '127.0.0.1' # 服务器IP地址
PORT = 8986 # 服务器端口号
# 创建socket对象
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 连接服务器
s.connect((HOST, PORT))
print(f"Connected to {HOST}:{PORT}")
# 发送数据
data = "Hello, Server"
msg = NewMessage(1, len(data), data.encode("utf-8"))
dp = NewDataPack()
send_data = dp.Pack(msg)
s.sendall(send_data)
# 接收响应数据
response = s.recv(1024)
print("Received from server: ", response)
# 关闭连接
s.close()
再做一个服务端。
# -*- coding: utf-8 -*-
import sys
sys.path.append("../../")
from znet.server import NewServer
from znet.router import BaseRouter
from ziface.irequest import IRequest
class PingRouter(BaseRouter):
"""
用户自定义路由
"""
def __init__(self):
super().__init__()
def Handle(self, request: IRequest):
"""
处理conn业务的方法
:param request:
:return:
"""
try:
print("调用Handle")
request.GetConnection().GetTCPConnection().send("ping\n".encode("GB2312"))
request.GetConnection().GetTCPConnection().send(request.GetData())
except Exception as e:
print("ping异常", e)
if __name__ == '__main__':
server = NewServer()
server.AddRouter(PingRouter())
server.Serve()
此时发送和接收都正常。消息封装模块完成。
