main.py 4.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121
  1. import time
  2. import signal # 新增:导入signal模块
  3. from multiprocessing import Process, cpu_count
  4. from typing import Dict
  5. from core.utils.log.logger_manager import LoggerManager
  6. from scheduler.process_manager import split_topics, start_worker_process
  7. from spiders.spider_registry import SPIDER_CLASS_MAP
  8. # 全局关闭标志
  9. shutdown_flag = False
  10. def handle_signal(sig, frame):
  11. """信号处理函数"""
  12. global shutdown_flag
  13. logger = LoggerManager.get_logger()
  14. logger.warning(f"[主进程] 接收到信号 {sig},开始优雅关闭...")
  15. shutdown_flag = True
  16. def main():
  17. global shutdown_flag
  18. logger = LoggerManager.get_logger()
  19. aliyun_log = LoggerManager.get_aliyun_logger()
  20. # 注册信号处理器 - 关键新增
  21. signal.signal(signal.SIGTERM, handle_signal) # 处理 kill
  22. signal.signal(signal.SIGINT, handle_signal) # 处理 Ctrl+C
  23. signal.signal(signal.SIGHUP, handle_signal) # 处理终端断开
  24. logger.info(f"[主进程] 启动,PID={os.getpid()}") # 记录PID便于管理
  25. topic_list = list(SPIDER_CLASS_MAP.keys())
  26. logger.info(f"[主进程] 监听 Topics: {topic_list}")
  27. max_processes = cpu_count()
  28. num_processes = min(len(topic_list), max_processes)
  29. topic_groups = split_topics(topic_list, num_processes)
  30. logger.info(f"[主进程] CPU 核心数: {max_processes},启动进程数: {num_processes}")
  31. process_map: Dict[int, Process] = {}
  32. for group_id, topic_group in enumerate(topic_groups):
  33. start_worker_process(group_id, topic_group, process_map)
  34. # 重启状态跟踪 - 新增重启限制
  35. restart_count = {group_id: 0 for group_id in range(len(topic_groups))}
  36. last_restart_time = {group_id: 0 for group_id in range(len(topic_groups))}
  37. # 主进程持续监控子进程状态
  38. try:
  39. while not shutdown_flag: # 检查关闭标志
  40. time.sleep(5)
  41. for group_id, p in list(process_map.items()):
  42. if shutdown_flag: # 如果正在关闭,跳过检查
  43. break
  44. if not p.is_alive():
  45. current_time = time.time()
  46. # 重启限制逻辑 - 防止无限重启
  47. if restart_count[group_id] > 5 and (current_time - last_restart_time[group_id]) < 60:
  48. logger.error(f"[监控] 进程 {p.name} 频繁崩溃,停止重启!")
  49. aliyun_log.logging(
  50. code="1502",
  51. message="进程频繁崩溃已停止重启",
  52. data={
  53. "pid": p.pid,
  54. "exitcode": p.exitcode,
  55. "restart_count": restart_count[group_id],
  56. "group_id": group_id,
  57. "topics": topic_groups[group_id]
  58. }
  59. )
  60. continue
  61. logger.warning(f"[监控] 进程 {p.name} PID={p.pid} 已崩溃,正在重启...")
  62. aliyun_log.logging(
  63. code="1501",
  64. message=f"[主进程监控] 子进程 {p.name} 崩溃重启",
  65. data={
  66. "pid": p.pid,
  67. "exitcode": p.exitcode,
  68. "group_id": group_id,
  69. "topics": topic_groups[group_id],
  70. "restart_count": restart_count[group_id] + 1
  71. }
  72. )
  73. # 更新重启状态
  74. restart_count[group_id] += 1
  75. last_restart_time[group_id] = current_time
  76. time.sleep(2)
  77. start_worker_process(group_id, topic_groups[group_id], process_map)
  78. except KeyboardInterrupt:
  79. logger.warning("[主进程] 接收到退出信号,终止所有子进程...")
  80. shutdown_flag = True
  81. # 优雅终止所有子进程 - 统一处理关闭
  82. logger.info("[主进程] 终止所有子进程...")
  83. for p in process_map.values():
  84. if p.is_alive():
  85. p.terminate() # 先尝试正常终止
  86. # 等待子进程退出
  87. for p in process_map.values():
  88. p.join(timeout=5) # 最多等待5秒
  89. if p.is_alive():
  90. logger.warning(f"进程 {p.pid} 未正常退出,强制终止")
  91. p.kill() # 强制终止
  92. p.join()
  93. logger.info("[主进程] 所有子进程已终止,退出")
  94. if __name__ == '__main__':
  95. import os
  96. main()