162 lines
5.7 KiB
Python
162 lines
5.7 KiB
Python
import socket
|
||
import signal
|
||
import threading
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
|
||
from config.setting import settings
|
||
from model.index import UNKNOWN_MESSAGE
|
||
from processor.tcp_processor import TcpMessageProcessor
|
||
from config.logger_config import logger
|
||
|
||
|
||
class TCPServer:
|
||
def __init__(self):
|
||
self.host = settings.host
|
||
self.port = settings.port
|
||
|
||
self.max_workers = settings.max_workers
|
||
self.timeout = settings.timeout
|
||
|
||
self.server_socket = None
|
||
self.running = False
|
||
self.thread_pool = ThreadPoolExecutor(max_workers=self.max_workers)
|
||
self.message_processor = TcpMessageProcessor()
|
||
|
||
# 已经连接的客户端
|
||
self.clients = {}
|
||
self.client_lock = threading.Lock()
|
||
|
||
self.server_thread = None
|
||
|
||
signal.signal(signal.SIGTERM, self._handle_signal)
|
||
signal.signal(signal.SIGINT, self._handle_signal)
|
||
|
||
def _handle_signal(self, signum, frame):
|
||
"""处理终止信号，触发优雅关闭"""
|
||
logger.info(f"收到信号 {signum}，准备关闭服务器...")
|
||
self.running = False
|
||
|
||
def _start_loop(self):
|
||
"""启动服务器"""
|
||
try:
|
||
self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||
self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||
self.server_socket.bind((self.host, self.port))
|
||
self.server_socket.listen(5)
|
||
self.server_socket.settimeout(1.0)
|
||
self.running = True
|
||
|
||
logger.info(f"TCP服务器启动，监听 {self.host}:{self.port} "
|
||
f"（最大线程: {self.max_workers}, 超时: {self.timeout}s）")
|
||
|
||
while self.running:
|
||
try:
|
||
client_socket, client_address = self.server_socket.accept()
|
||
client_ip = client_address[0]
|
||
client_socket.settimeout(self.timeout)
|
||
logger.info(f"新连接: {client_address}")
|
||
|
||
# 存储客户端
|
||
with self.client_lock:
|
||
self.clients[client_ip] = client_socket
|
||
|
||
# 提交到线程池处理
|
||
self.thread_pool.submit(self.handle_client, client_socket, client_ip)
|
||
except socket.timeout:
|
||
continue
|
||
except Exception as e:
|
||
if self.running:
|
||
logger.error(f"接受连接失败: {str(e)}")
|
||
except Exception as e:
|
||
logger.error(f"TCP服务器启动失败: {str(e)}")
|
||
finally:
|
||
self.stop()
|
||
|
||
def start(self):
|
||
self.server_thread = threading.Thread(target=self._start_loop, daemon=True)
|
||
self.server_thread.start()
|
||
|
||
def stop(self):
|
||
if not self.running:
|
||
return
|
||
|
||
self.running = False
|
||
logger.info("开始关闭服务器...")
|
||
|
||
# 移除客户端
|
||
with self.client_lock:
|
||
for client_socket in self.clients.values():
|
||
try:
|
||
client_socket.close()
|
||
except Exception as e:
|
||
logger.warning(f"关闭客户端连接失败: {str(e)}")
|
||
self.clients.clear()
|
||
|
||
# 关闭线程池
|
||
self.thread_pool.shutdown(wait=True)
|
||
logger.info("所有客户端处理线程已结束")
|
||
|
||
# 关闭连接
|
||
if self.server_socket:
|
||
self.server_socket.close()
|
||
logger.info(f"服务器已关闭（{self.host}:{self.port}）")
|
||
|
||
def handle_client(self, client_socket, client_ip):
|
||
"""处理客户端连接"""
|
||
try:
|
||
while True:
|
||
data = client_socket.recv(1024)
|
||
if not data:
|
||
logger.info(f"客户端 {client_ip} 主动断开连接")
|
||
break
|
||
|
||
message = data.decode('utf-8').strip()
|
||
logger.info(f"收到 {client_ip} 的消息: {message}")
|
||
|
||
# 放到消息处理器里面处理
|
||
response = self.message_processor.process(message)
|
||
|
||
# 回复消息
|
||
if response != UNKNOWN_MESSAGE:
|
||
client_socket.sendall(response.encode('utf-8'))
|
||
logger.info(f"回复 {client_ip}: {response}")
|
||
except socket.timeout:
|
||
logger.warning(f"客户端 {client_ip} 超时未活动")
|
||
except Exception as e:
|
||
logger.error(f"处理 {client_ip} 出错: {str(e)}")
|
||
finally:
|
||
# 异常情况下关闭连接
|
||
with self.client_lock:
|
||
if client_ip in self.clients:
|
||
del self.clients[client_ip]
|
||
|
||
try:
|
||
client_socket.close()
|
||
logger.info(f"客户端 {client_ip} 连接已关闭")
|
||
except Exception as e:
|
||
logger.warning(f"关闭 {client_ip} 连接失败: {str(e)}")
|
||
|
||
def send_to_client(self, client_ip, message):
|
||
# 先获取客户端连接（加锁保护）
|
||
with self.client_lock:
|
||
client_socket = self.clients.get(client_ip)
|
||
if not client_socket:
|
||
logger.warning(f"客户端 {client_ip} 不存在或已断开连接")
|
||
return False
|
||
|
||
# 发送消息
|
||
try:
|
||
client_socket.sendall(message.encode('utf-8'))
|
||
logger.info(f"主动发送消息给 {client_ip}: {message}")
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"向 {client_ip} 发送消息失败: {str(e)}")
|
||
# 发送失败时移除无效连接
|
||
with self.client_lock:
|
||
if client_ip in self.clients:
|
||
del self.clients[client_ip]
|
||
return False
|
||
|
||
|
||
tcp_server = TCPServer()
|