修复linux下执行pm2 restart后python守护进程未重启

修复linux下执行pm2 restart后python守护进程未重启

Author: sanshu-rom

spider/py/core/t4_daemon.py

@@ -27,11 +27,11 @@
 INIT_TIMEOUT = 15  # 初始化超时（秒）
 IDLE_EXPIRE = 30 * 60  # 实例空闲过期（秒）
 CLEAN_INTERVAL = 5 * 60  # 清理间隔（秒）
-REQUEST_TIMEOUT = 120  # 单次请求socket超时（秒）
+REQUEST_TIMEOUT = 120  # 单次请求 socket 超时（秒）
 
 LOG_LEVEL = os.environ.get("T4_LOG_LEVEL", "INFO").upper()
 LOG_FILE = os.environ.get("T4_LOG_FILE")  # 若未设置则只打到控制台
-PID_FILE = os.environ.get("T4_PID_FILE")  # 若设置则会写入PID
+PID_FILE = os.environ.get("T4_PID_FILE")  # 若设置则会写入 PID
 
 # =========================
 # 日志
@@ -71,7 +71,6 @@
     'action': 'action',
 }
 
-
 # =========================
 # 工具：长度前缀协议
 # =========================
@@ -87,7 +86,6 @@ def recv_exact(rfile, n: int) -> bytes:
         remaining -= len(chunk)
     return b"".join(chunks)
 
-
 def send_packet(wfile, obj: dict):
     payload = pickle.dumps(obj, protocol=pickle.HIGHEST_PROTOCOL)
     if len(payload) > MAX_MSG_SIZE:
@@ -96,7 +94,6 @@ def send_packet(wfile, obj: dict):
     wfile.write(payload)
     wfile.flush()
 
-
 def recv_packet(rfile) -> dict:
     header = recv_exact(rfile, 4)
     (length,) = struct.unpack(">I", header)
@@ -105,7 +102,6 @@ def recv_packet(rfile) -> dict:
     payload = recv_exact(rfile, length)
     return pickle.loads(payload)
 
-
 # =========================
 # Spider 管理
 # =========================
@@ -117,7 +113,6 @@ def __init__(self, spider):
         self.init_event = threading.Event()
         self.last_used = time.time()
 
-
 class SpiderManager:
     def __init__(self, logger):
         self.logger = logger
@@ -154,7 +149,6 @@ def _parse_env(env_str: str):
                 proxy_url = data.get("proxyUrl", "") or ""
                 ext = data.get("ext", "") or ""
             except Exception:
-                # 非JSON字符串时，保持兼容：当作 ext 传
                 ext = env_str
         return proxy_url, ext
 
@@ -163,12 +157,10 @@ def _instance_key(self, script_path: str, env_str: str) -> str:
         key_data = f"{Path(script_path).resolve()}|{proxy_url}|{ext}"
         return hashlib.sha256(key_data.encode("utf-8")).hexdigest()
 
-    # ---------- 动态导入 ----------
     def _load_module_from_file(self, file_path: Path):
         name = file_path.stem
         logger.info("_load_module_from_file %s", name)
-        # 加入项目根目录到 sys.path，保证 base.* 可以被导入
-        project_root = file_path.parent  # 假设 py 是根目录
+        project_root = file_path.parent
         if str(project_root) not in sys.path:
             sys.path.insert(0, str(project_root))
             logger.info("Added %s to sys.path", project_root)
@@ -184,7 +176,6 @@ def _import_spider_module(self, script_path: str):
         p = Path(script_path)
         if p.exists() and p.is_file() and p.suffix == ".py":
             return self._load_module_from_file(p)
-        # 作为模块路径导入（已在 sys.path 中）
         return importlib.import_module(script_path)
 
     def _create_spider(self, script_path: str, env_str: str):
@@ -237,11 +228,9 @@ def _ensure_instance(self, script_path: str, env_str: str) -> SpiderInstance:
             return inst
 
     def call(self, script_path: str, method_name: str, env_str: str, args_list):
-        # 解析 env 中 ext
         _, ext = self._parse_env(env_str)
         inst = self._ensure_instance(script_path, env_str)
 
-        # init 分支：同步初始化
         if method_name == "init":
             with threading.Lock():
                 if inst.initializing:
@@ -258,7 +247,6 @@ def call(self, script_path: str, method_name: str, env_str: str, args_list):
                         inst.initializing = False
                 return {"status": "already initialized"}
 
-        # 其他方法：若未初始化，则异步触发 + 等待
         if not inst.initialized:
             if not inst.initializing:
                 def _bg():
@@ -267,16 +255,12 @@ def _bg():
                         inst.initialized = True
                         inst.init_event.set()
                     except Exception:
-                        # 失败也置事件，避免永等
                         inst.init_event.set()
-
                 inst.initializing = True
                 threading.Thread(target=_bg, daemon=True).start()
-
             if not inst.init_event.wait(INIT_TIMEOUT) or not inst.initialized:
                 return {"success": False, "error": "init timeout or failed"}
 
-        # 解析 args
         parsed_args = []
         for a in (args_list or []):
             if isinstance(a, (dict, list, int, float, bool, type(None))):
@@ -289,14 +273,12 @@ def _bg():
             else:
                 parsed_args.append(a)
 
-        # 方法映射
         invoke = METHOD_MAP.get(method_name, method_name)
         if not hasattr(inst.spider, invoke):
             return {"success": False, "error": f"Spider missing method '{invoke}'"}
 
         try:
             result = getattr(inst.spider, invoke)(*parsed_args)
-            # 若 Spider 提供 json2str 则尝试序列化
             if result is not None and hasattr(inst.spider, "json2str"):
                 try:
                     return inst.spider.json2str(result)
@@ -311,13 +293,11 @@ def _bg():
                 "traceback": traceback.format_exc(),
             }
 
-
 # =========================
 # Server
 # =========================
 _manager = SpiderManager(logger)
 
-
 class T4Handler(StreamRequestHandler):
     def handle(self):
         self.request.settimeout(REQUEST_TIMEOUT)
@@ -337,41 +317,50 @@ def handle(self):
                 resp["error"] = result.get("error")
                 if result.get("traceback"):
                     resp["traceback"] = result["traceback"]
-
             send_packet(self.wfile, resp)
         except Exception as e:
             logger.error("Handle error: %s", e)
             try:
                 send_packet(self.wfile, {"success": False, "error": str(e)})
             except Exception:
-                pass  # 对端已断开
-
+                pass
 
 class ThreadedTCPServer(ThreadingMixIn, TCPServer):
     daemon_threads = True
     allow_reuse_address = True
 
-
 def run():
-    def _stop(*_):
-        logger.info("Stopping server ...")
+    # ✅ 让当前进程成为进程组组长，方便整组 kill
+    if os.name == "posix":
+        os.setpgrp()
+
+    def _stop(signum, frame):
+        """
+        收到 SIGINT/SIGTERM 时的回调：
+        1. 停止 SpiderManager
+        2. 关闭 TCP Server
+        3. 强杀整个进程组，确保进程立即退出
+        """
+        logger.info("Received %s, shutting down...", signum)
         _manager.stop()
-        # 让 serve_forever() 退出
         srv.shutdown()
+        # 立即强制退出，防止线程阻塞
+        os._exit(0)
 
+    # 注册信号
     if os.name == "posix":
-        signal.signal(signal.SIGTERM, _stop)
         signal.signal(signal.SIGINT, _stop)
+        signal.signal(signal.SIGTERM, _stop)
 
     global srv
     srv = ThreadedTCPServer((HOST, PORT), T4Handler)
     logger.info("T4 daemon listening on %s:%d", HOST, PORT)
+
     try:
         srv.serve_forever(poll_interval=0.5)
     finally:
         srv.server_close()
         logger.info("Server closed.")
 
-
 if __name__ == "__main__":
     run()