美文网首页
Python版本Zinx——(5)消息封装(封包、拆包)

Python版本Zinx——(5)消息封装(封包、拆包)

作者: 爱折腾的胖子 | 来源:发表于2023-08-27 00:34 被阅读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()
    
    

      此时发送和接收都正常。消息封装模块完成。

    相关文章

      网友评论

          本文标题:Python版本Zinx——(5)消息封装(封包、拆包)

          本文链接:https://www.haomeiwen.com/subject/bcnymdtx.html