最近想研究一下关于长链接的相关内容，在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)和第三方库来实现类似于接口的功能。在实际开发中，我们可以根据具体需求选择合适的实现方式。
暂时使用抽象基类的形式模拟接口的实现。
上一节handlerAPI写的过于潦草了，本节就封装一下。
首先封装链接和数据。在ziface中创建一个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
在znet中创建request.py，做IRequest的实现类。
# -*- coding: utf-8 -*-
from ziface.irequest import IRequest
from ziface.iconnection import IConnection
class Request(IRequest):
def __init__(self, conn: IConnection, data: bytes):
"""
:param conn: 已经和客户端建立好的 链接
:param msg: 客户端请求的数据
"""
self.Conn: IConnection = conn
self.Data: bytes = data
def GetConnection(self) -> IConnection:
"""
获取请求连接信息
:return:
"""
return self.Conn
def GetData(self) -> bytes:
"""
获取请求消息的数据
:return:
"""
return self.Data
def NewRequest(conn: IConnection, data: bytes) -> IRequest:
req = Request(conn, data)
return req
链接和数据封装好了之后，做一下路由模块的基础封装。在ziface中创建一个irouter。
# -*- coding: utf-8 -*-
from ziface.irequest import IRequest
from abc import ABC, abstractmethod
class IRouter(ABC):
"""
路由接口， 这里面路由是 使用框架者给该链接自定的 处理业务方法
路由里的IRequest 则包含用该链接的链接信息和该链接的请求数据信息
"""
@abstractmethod
def PreHandle(self, request: IRequest):
"""
在处理conn业务之前的钩子方法
:param request:
:return:
"""
pass
@abstractmethod
def Handle(self, request: IRequest):
"""
处理conn业务的方法
:param request:
:return:
"""
pass
@abstractmethod
def PostHandle(self, request: IRequest):
"""
处理conn业务之后的钩子方法
:param request:
:return:
"""
pass
在znet中创建router.py，做IRouter的实现类，这个需要做成一个基类，用户自定义的路由模块直接继承基类就行了。
# -*- coding: utf-8 -*-
from ziface.irequest import IRequest
from ziface.irouter import IRouter
class BaseRouter(IRouter):
"""
实现router时，先嵌入这个基类，然后根据需要对这个基类的方法进行重写
这里之所以BaseRouter的方法都为空，是因为有的Router不希望有PreHandle或PostHandle
所以Router全部继承BaseRouter的好处是，不需要实现PreHandle和PostHandle也可以实例化
"""
def __init__(self):
pass
def PreHandle(self, request: IRequest):
"""
在处理conn业务之前的钩子方法
:param request:
:return:
"""
pass
def Handle(self, request: IRequest):
"""
处理conn业务的方法
:param request:
:return:
"""
pass
def PostHandle(self, request: IRequest):
"""
处理conn业务之后的钩子方法
:param request:
:return:
"""
pass
Request模块与Router模块封装好了之后，在Server和Connection中集成。
在IServer新增一个添加路由的方法。
# -*- coding: utf-8 -*-
from ziface.irouter import IRouter
from abc import ABC, abstractmethod
class IServer(ABC):
"""
定义服务器接口
"""
@abstractmethod
def Start(self):
"""
启动服务器方法
:return:
"""
pass
@abstractmethod
def Stop(self):
"""
停止服务器方法
:return:
"""
pass
@abstractmethod
def Serve(self):
"""
开启业务服务方法
:return:
"""
pass
@abstractmethod
def AddRouter(self, router: IRouter):
"""
添加路由
:param router:
:return:
"""
pass
将Connection中的handlerAPI删除，添加一个Router，并且在StartReader中添加响应的处理Router的代码。
# -*- coding: utf-8 -*-
import socket
import gevent
from ziface.iconnection import IConnection
from ziface.irequest import IRequest
from ziface.irouter import IRouter
from znet.request import NewRequest
# 存一下协程
GlobalGevents: list = []
class Connection(IConnection):
def __init__(self, conn: socket.socket, connID: int, remote_addr: tuple, router:IRouter):
self.Conn: socket.socket = conn # 当前链接的socket TCP套接字
self.ConnID: int = connID # 链接的ID
# self.HandlerAPI = handlerAPI # 当前链接绑定的业务处理方法的API
self.is_closed: bool = False # 链接状态
self.Remote_Addr: tuple = remote_addr
self.Router: IRouter = router
def Start(self):
"""
启动链接 让当前的链接准备开始工作
:return:
"""
print("链接开启，ID=", self.ConnID)
# 开启读业务
g1 = gevent.spawn(self.StartReader)
GlobalGevents.append(g1)
def StartReader(self):
"""
处理读业务
:return:
"""
print("开启读业务")
while True:
try:
# 读取客户端的Msg head
read_buf = self.Conn.recv(512)
if len(read_buf) == 0:
# head_data 长度为0表示客户端已经退出
break
# self.HandlerAPI(self.Conn, read_buf)
req = NewRequest(self, read_buf)
g2 = gevent.spawn(self.RunHandler(req))
GlobalGevents.append(g2)
except Exception as e:
print("读取数据异常 ", e)
break
print("读业务关闭，ID", self.ConnID)
self.Stop()
def RunHandler(self, request: IRequest):
self.Router.PreHandle(request)
self.Router.Handle(request)
self.Router.PostHandle(request)
def Stop(self):
"""
停止链接 结束当前链接的工作
:return:
"""
print("链接关闭，ID=", self.ConnID)
if self.is_closed:
return
self.is_closed = True
# 关闭socket链接，回收资源
self.Conn.close()
def GetTCPConnection(self) -> socket.socket:
"""
获取当前链接绑定的socket conn
:return:
"""
return self.Conn
def GetConnID(self) -> int:
"""
获取当前链接模块的链接ID
:return:
"""
return self.ConnID
def GetRemoteAddr(self) -> tuple:
"""
获取远程客户端的TCP状态 IP Port
:return:
"""
return self.Remote_Addr
def NewConnection(conn: socket.socket, connID: int, remote_addr: tuple, router: IRouter) -> IConnection:
c = Connection(conn, connID, remote_addr, router)
return c
将Server中的CallBackToClient删除，添加一个Router，并且在InitTCP中添加响应的处理Router的代码。
# -*- coding: utf-8 -*-
from typing import Optional
from ziface.irouter import IRouter
from ziface.iserver import IServer
from znet.connection import NewConnection
from gevent import monkey
import time
import socket
import gevent
import signal
import sys
# 把内置阻塞的库替换成非阻塞的socket
monkey.patch_socket()
# 存一下协程
GlobalGevents: list = []
class Server(IServer):
def __init__(self, name: str, ip: str, port: int,
family: socket.AddressFamily = socket.AF_INET,
socket_kind: socket.SocketKind = socket.SOCK_STREAM):
self.name: str = name
self.family: socket.AddressFamily = family
self.socket_kind: socket.SocketKind = socket_kind
self.ip: str = ip
self.port: int = port
self.Router: Optional[IRouter] = None
def Start(self):
"""
启动服务器方法
:return:
"""
print("[启动] 服务监听，IP地址：%s, 端口：%s，已经启动\n" % (self.ip, self.port))
g1 = gevent.spawn(self.InitTCP)
GlobalGevents.append(g1)
def InitTCP(self):
"""
初始化TCP链接
:return:
"""
# 1.获取一个TCP的Addr
with socket.socket(self.family, self.socket_kind) as tcp:
# 2.监听服务器的地址
tcp.bind((self.ip, self.port))
tcp.listen(128)
print("[启动] %s服务启动成功，监听中......" % self.name)
# 3.阻塞的等待客户端链接，处理客户端链接业务（读写）。
cid: int = 0
while True:
print("开启接收")
remote_socket, remote_addr = tcp.accept()
dealConn = NewConnection(remote_socket, cid, remote_addr, self.Router)
cid += 1
g2 = gevent.spawn(dealConn.Start())
GlobalGevents.append(g2)
def Stop(self):
"""
停止服务器方法
:return:
"""
print("服务停止")
def Serve(self):
"""
开启业务服务方法
:return:
"""
self.Start()
gevent.joinall(GlobalGevents)
# 阻塞一下
while True:
time.sleep(10)
def AddRouter(self, router: IRouter):
"""
添加路由
:param router:
:return:
"""
self.Router = router
print("路由添加成功")
# def CallBackToClient(conn: socket.socket, data: bytes):
# print("回显功能")
# try:
# conn.send(data)
# except Exception as e:
# print(e)
# raise Exception("回显功能异常")
# 创建一个全局的server
m_server: Optional[IServer] = None
# 获取server
def NewServer() -> IServer:
global m_server
if m_server:
return m_server
m_server = Server("测试", "0.0.0.0", 8986)
return m_server
# 接收停止信号，要不然报错看着烦
def signal_handler(signal, frame):
global m_server
m_server.Stop()
print("程序已退出")
sys.exit(0)
# 接收停止信号 ctrl+c
signal.signal(signal.SIGINT, signal_handler)
创建一个demo文件夹，用于写测试方法。demo中创建router文件夹，表明是测试Router。router文件夹中，创建一个root.py，写上测试函数。
# -*- 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 PreHandle(self, request: IRequest):
"""
在处理conn业务之前的钩子方法
:param request:
:return:
"""
try:
print("调用PreHandle")
request.GetConnection().GetTCPConnection().send("ping前\n".encode("GB2312"))
except Exception as e:
print("ping前异常", e)
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)
def PostHandle(self, request: IRequest):
"""
处理conn业务之后的钩子方法
:param request:
:return:
"""
try:
print("调用PostHandle")
request.GetConnection().GetTCPConnection().send("ping后\n".encode("GB2312"))
except Exception as e:
print("ping后异常", e)
if __name__ == '__main__':
server = NewServer()
server.AddRouter(PingRouter())
server.Serve()
服务接口做完了。在cmd中输入命令，回车
nc 127.0.0.1 8986
输入几个字符，回车键发送，此时封装好的Router和Request已经集成到Zinx中，用户自定义的PingRouter回显功能可以正常运行。
