1 源码
本服务未使用第三方库,所有端口均可以在配置文件中新增、删除、和修改。
import socketserver
import threading
import time
import queue
import json
__version__ = "1.0.0"
__author__ = "madi"
__id__ = "smile"
__all__ = ["Main",]
"""
版权所有,供开发测试使用,禁止商用,违者必究
"""
# 透传基类,Main动态继承
class BaseTrans(socketserver.BaseRequestHandler):
write: queue.Queue = None
read: queue.Queue= None
def __init__(self,*args,**kwargs):
super(BaseTrans, self).__init__(*args,**kwargs)
# 接收到的字节流,外部可以访问
self.content = None
def handle(self):
# 连接客户端的对象
self._conn = self.request
# 开启发送线程
threading.Thread(target=self.send_).start()
# 进入业务主循环
while True:
try:
self.content = self._conn.recv(1000000)
if not len(self.content):
del self._conn
break
self.write.put(self.content)
except ConnectionResetError: break
except UnboundLocalError: break
def send_(self):
while True:
time.sleep(0.005)
try:
if not self.read.empty():
self._conn.send(self.read.get())
except: break
# 超级列表类,额,开个玩笑,仅仅是链式调用
class SuperList(list):
def __init__(self, *args, **kwargs):
super(SuperList, self).__init__(*args, **kwargs)
def add(self, content):
self.append(content)
return self
class Main:
def __init__(self):
# 创建BaseTrans子类对象池
self.Master_pool = SuperList()
# 创建消息队列对象池
self.Queue_pool = SuperList()
# 获取配置文件信息
with open("./config.json", "r", encoding="utf-8") as file_:
self.config_data = json.load(file_)
# 计算出须创建的池的大小
self.pool_length = len(self.config_data)
# 配置文件中端口池
self.port_pool = []
# 对象生成器
def create_class(self):
for i in range(self.pool_length):
# 创建消息队列
# 创建写消息队列
self.Queue_pool\
.add(queue.Queue())\
.add(queue.Queue())
# 动态创建BaseTrans子类对象,放入池中
self.Master_pool\
.add(
# 动态创建BaseTrans子类对象
type("Master{}".format(2 * i), (BaseTrans,), {
"write": self.Queue_pool[2 * i], "read":self.Queue_pool[2 * i + 1]})
)\
.add(
type("Master{}".format(2 * i + 1), (BaseTrans,), {
"write": self.Queue_pool[2 * i + 1], "read": self.Queue_pool[2 * i]})
)
# 加入配置文件中的端口
self.port_pool.extend(self.config_data[str(i)])
@staticmethod
def open(ip, port, socket_):
# 开启监听线程
server_ = socketserver.ThreadingTCPServer((ip, port), socket_)
server_.serve_forever()
def run(self):
self.create_class()
# 创建动作线程
for i in range(self.pool_length * 2):
threading.Thread(target=self.open, args=("0.0.0.0", self.port_pool[i], self.Master_pool[i])).start()
if __name__ == '__main__':
my_server = Main()
my_server.run()
2 源码下载地址
Gitee下载:扫地僧-smile
使用方法在README中有介绍,可以点击加关注哦!!!