209 lines
7.1 KiB
Python
209 lines
7.1 KiB
Python
import subprocess
|
|
import time
|
|
import psutil
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
import threading
|
|
import logging
|
|
|
|
|
|
class GameProcessManager:
|
|
def __init__(self, exe_path, max_processes=24, timeout_minutes=30):
|
|
"""
|
|
初始化游戏进程管理器
|
|
|
|
Args:
|
|
exe_path: 游戏 exe 文件的完整路径
|
|
max_processes: 最大同时运行的游戏进程数
|
|
timeout_minutes: 单个游戏进程的超时时间(分钟)
|
|
"""
|
|
self.exe_path = Path(exe_path)
|
|
self.max_processes = max_processes
|
|
self.timeout_seconds = timeout_minutes * 60
|
|
self.processes = {} # {进程对象: 启动时间}
|
|
self.running = True
|
|
|
|
# 配置日志
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(levelname)s - %(message)s',
|
|
handlers=[
|
|
logging.FileHandler('game_manager.log', encoding='utf-8'),
|
|
logging.StreamHandler()
|
|
]
|
|
)
|
|
self.logger = logging.getLogger(__name__)
|
|
|
|
# 验证 exe 文件存在
|
|
if not self.exe_path.exists():
|
|
raise FileNotFoundError(f"游戏 exe 文件不存在: {self.exe_path}")
|
|
|
|
def start_game_process(self):
|
|
"""启动一个新的游戏进程"""
|
|
try:
|
|
process = subprocess.Popen(
|
|
str(self.exe_path),
|
|
cwd=str(self.exe_path.parent),
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL
|
|
)
|
|
self.processes[process] = datetime.now()
|
|
self.logger.info(f"启动新游戏进程 PID: {process.pid}, 当前进程数: {len(self.processes)}")
|
|
return process
|
|
except Exception as e:
|
|
self.logger.error(f"启动游戏进程失败: {e}")
|
|
return None
|
|
|
|
def check_and_kill_timeout_processes(self):
|
|
"""检查并关闭超时的游戏进程"""
|
|
current_time = datetime.now()
|
|
timeout_processes = []
|
|
|
|
for process, start_time in list(self.processes.items()):
|
|
elapsed_time = (current_time - start_time).total_seconds()
|
|
|
|
if elapsed_time > self.timeout_seconds:
|
|
timeout_processes.append(process)
|
|
|
|
for process in timeout_processes:
|
|
try:
|
|
# 尝试优雅地终止进程
|
|
process.terminate()
|
|
time.sleep(2)
|
|
|
|
# 如果进程还在运行,强制杀死
|
|
if process.poll() is None:
|
|
process.kill()
|
|
|
|
elapsed = (current_time - self.processes[process]).total_seconds() / 60
|
|
self.logger.warning(f"强制关闭超时进程 PID: {process.pid}, 运行时间: {elapsed:.1f} 分钟")
|
|
del self.processes[process]
|
|
except Exception as e:
|
|
self.logger.error(f"关闭超时进程失败 PID: {process.pid}, 错误: {e}")
|
|
|
|
def clean_finished_processes(self):
|
|
"""清理已完成的游戏进程"""
|
|
finished_processes = []
|
|
|
|
for process in list(self.processes.keys()):
|
|
if process.poll() is not None:
|
|
finished_processes.append(process)
|
|
|
|
for process in finished_processes:
|
|
elapsed = (datetime.now() - self.processes[process]).total_seconds() / 60
|
|
self.logger.info(f"游戏进程正常退出 PID: {process.pid}, 运行时间: {elapsed:.1f} 分钟")
|
|
del self.processes[process]
|
|
|
|
return len(finished_processes)
|
|
|
|
def get_active_process_count(self):
|
|
"""获取当前活跃的游戏进程数"""
|
|
return len(self.processes)
|
|
|
|
def maintain_process_pool(self):
|
|
"""维护进程池,确保进程数量达到设定值"""
|
|
# 清理已完成的进程
|
|
finished_count = self.clean_finished_processes()
|
|
|
|
# 检查并关闭超时进程
|
|
self.check_and_kill_timeout_processes()
|
|
|
|
# 计算需要启动的进程数
|
|
current_count = self.get_active_process_count()
|
|
needed_count = self.max_processes - current_count
|
|
|
|
# 启动新进程补充到目标数量
|
|
if needed_count > 0:
|
|
self.logger.info(f"需要启动 {needed_count} 个新进程 (当前: {current_count}/{self.max_processes})")
|
|
for _ in range(needed_count):
|
|
self.start_game_process()
|
|
time.sleep(1) # 避免同时启动太多进程
|
|
|
|
def run(self):
|
|
"""主循环:持续监控和维护游戏进程"""
|
|
self.logger.info(f"游戏进程管理器启动")
|
|
self.logger.info(f"游戏路径: {self.exe_path}")
|
|
self.logger.info(f"最大进程数: {self.max_processes}")
|
|
self.logger.info(f"超时时间: {self.timeout_seconds / 60} 分钟")
|
|
|
|
try:
|
|
while self.running:
|
|
self.maintain_process_pool()
|
|
time.sleep(5) # 每5秒检查一次
|
|
except KeyboardInterrupt:
|
|
self.logger.info("接收到停止信号,正在关闭所有游戏进程...")
|
|
self.stop()
|
|
|
|
def stop(self):
|
|
"""停止管理器并关闭所有游戏进程"""
|
|
self.running = False
|
|
|
|
for process in list(self.processes.keys()):
|
|
try:
|
|
process.terminate()
|
|
time.sleep(1)
|
|
if process.poll() is None:
|
|
process.kill()
|
|
self.logger.info(f"关闭游戏进程 PID: {process.pid}")
|
|
except Exception as e:
|
|
self.logger.error(f"关闭进程失败 PID: {process.pid}, 错误: {e}")
|
|
|
|
self.processes.clear()
|
|
self.logger.info("所有游戏进程已关闭")
|
|
|
|
def get_status(self):
|
|
"""获取当前状态信息"""
|
|
status = {
|
|
'total_processes': len(self.processes),
|
|
'max_processes': self.max_processes,
|
|
'processes_info': []
|
|
}
|
|
|
|
current_time = datetime.now()
|
|
for process, start_time in self.processes.items():
|
|
elapsed_minutes = (current_time - start_time).total_seconds() / 60
|
|
status['processes_info'].append({
|
|
'pid': process.pid,
|
|
'running_time_minutes': round(elapsed_minutes, 2)
|
|
})
|
|
|
|
return status
|
|
|
|
|
|
def main():
|
|
"""主函数"""
|
|
# ====== 配置参数 ======
|
|
# 修改为你的游戏 exe 路径
|
|
GAME_EXE_PATH = r"F:\pack\TOHOTOPIA Demo.exe"
|
|
|
|
# 最大同时运行的游戏数量
|
|
MAX_PROCESSES = 10
|
|
|
|
# 单个游戏超时时间(分钟)
|
|
TIMEOUT_MINUTES = 20
|
|
# =====================
|
|
|
|
manager = GameProcessManager(
|
|
exe_path=GAME_EXE_PATH,
|
|
max_processes=MAX_PROCESSES,
|
|
timeout_minutes=TIMEOUT_MINUTES
|
|
)
|
|
|
|
# 启动监控线程,定期输出状态
|
|
def status_monitor():
|
|
while manager.running:
|
|
time.sleep(60) # 每分钟输出一次状态
|
|
status = manager.get_status()
|
|
manager.logger.info(
|
|
f"状态报告 - 活跃进程: {status['total_processes']}/{status['max_processes']}"
|
|
)
|
|
|
|
monitor_thread = threading.Thread(target=status_monitor, daemon=True)
|
|
monitor_thread.start()
|
|
|
|
# 运行主循环
|
|
manager.run()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |