最近想研究一下关于长链接的相关内容，在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)和第三方库来实现类似于接口的功能。在实际开发中，我们可以根据具体需求选择合适的实现方式。
暂时使用抽象基类的形式模拟接口的实现。
第一节已经简单的完成了一个tcp服务的回显业务。本节继续完成tcp链接的封装。
首先是和iserver一样，在ziface中创建一个iconnection。
# -*- 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
在znet中创建connection.py文件，里面做一个IConnection的实现类。
# -*- coding: utf-8 -*-
from typing import Dict
from ziface.iconnection import IConnection
import socket
import gevent
# 存一下协程
GlobalGevents: list = []
class Connection(IConnection):
def __init__(self, conn: socket.socket, connID: int, remote_addr: tuple, handlerAPI):
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
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)
except Exception as e:
print("读取数据异常 ", e)
break
print("读业务关闭，ID", self.ConnID)
self.Stop()
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, handlerAPI) -> IConnection:
c = Connection(conn, connID, remote_addr, handlerAPI)
return c
封装好了Connection，将它整合进Server中。主要就是将回显功能提取出来，单独写了一个方法CallBackToClient，这个就是用户在使用的时候，可以对socket读取的数据自行设定处理方式。InitTCP函数中新建了connection，存放一个客户端的TCP链接。
# -*- coding: utf-8 -*-
from typing import Optional
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
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, CallBackToClient)
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 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)
服务接口做完了。在cmd中输入命令，回车
nc 127.0.0.1 8986
输入几个字符，回车键发送，此时封装好的Connection和用户自定义业务绑定就已经和Server一起运行了，用户自定义的回显功能可以正常运行。
网友评论