更新 README.md

src/infra/xxl_jobs/xxl_executor.py

@@ -22,6 +22,7 @@ logger = logging.getLogger(__name__)
 # Admin API 路径常量
 _PATH_REGISTRY = "/api/registry"
 _PATH_REGISTRY_REMOVE = "/api/registryRemove"
+_PATH_CALLBACK = "/api/callback"
 
 
 class XxlJobExecutor:
@@ -128,6 +129,46 @@ class XxlJobExecutor:
             except Exception:
                 pass
 
+    async def _callback(
+        self, log_id: int, log_date_time: int, result: dict
+    ) -> None:
+        """向 Admin 回调上报任务执行结果 —— POST /api/callback
+
+        对标 Java 版的 TriggerCallbackThread.callback()。
+
+        Args:
+            log_id: XXL-JOB 调度日志 ID
+            log_date_time: 调度时间戳（毫秒）
+            result: handler 返回的 {"code": 200, "msg": "..."} 或异常结果
+        """
+        if not log_id:
+            logger.warning("Skip callback: logId is empty")
+            return
+
+        body = {
+            "logId": log_id,
+            "logDateTim": log_date_time,
+            "handleCode": result.get("code", 500),
+            "handleMsg": result.get("msg", ""),
+        }
+        for admin_url in self._admin_urls():
+            try:
+                async with self._session.post(
+                    f"{admin_url}{_PATH_CALLBACK}",
+                    json=body,
+                    headers=self._auth_headers(),
+                ) as resp:
+                    if resp.status == 200:
+                        data = await resp.json()
+                        if data.get("code") == 200:
+                            return
+                        logger.warning(
+                            "Callback to %s rejected: %s", admin_url, data.get("msg")
+                        )
+            except Exception as e:
+                logger.warning("Callback to %s failed: %s", admin_url, e)
+        logger.error("All callback attempts failed for logId=%s", log_id)
+
     async def _heartbeat_loop(self) -> None:
         """定期心跳：每隔 registry_interval 秒重新注册（续约）"""
         while self._running:
@@ -154,18 +195,23 @@ class XxlJobExecutor:
     async def run(self, trigger_params: Dict[str, Any]) -> dict:
         """执行任务 —— Admin 调用 POST /run
 
+        XXL-JOB 标准行为：接收调度后立即返回，任务在后台执行，
+        执行完成后通过 POST /api/callback 向 Admin 上报结果。
+
         trigger_params 字段（XXL-JOB 标准）:
             jobId, executorHandler, executorParams, executorBlockStrategy,
             executorTimeout, logId, logDateTime, glueType, glueSource,
             glueUpdatetime, broadcastIndex, broadcastTotal
 
         Returns:
-            {"code": 200, "msg": None} 或 {"code": 500, "msg": "error detail"}
+            {"code": 200, "msg": null}  — 立即返回（不等待任务完成）
         """
         job_id = trigger_params.get("jobId")
         handler_name = trigger_params.get("executorHandler")
         param = trigger_params.get("executorParams", "")
         timeout = trigger_params.get("executorTimeout", 0)
+        log_id = trigger_params.get("logId", 0)
+        log_date_time = trigger_params.get("logDateTime", 0)
 
         if not handler_name:
             return {"code": 500, "msg": "executorHandler is required"}
@@ -179,31 +225,44 @@ class XxlJobExecutor:
                 f"Registered: {', '.join(registered) if registered else '(none)'}",
             }
 
-        # 异步执行，带超时
-        async def _run_with_timeout():
+        async def _execute_and_callback():
+            """后台执行 handler 并回调 Admin 上报结果"""
             try:
                 if timeout and timeout > 0:
-                    return await asyncio.wait_for(handler(param), timeout=timeout)
-                return await handler(param)
+                    result = await asyncio.wait_for(handler(param), timeout=timeout)
+                else:
+                    result = await handler(param)
             except asyncio.TimeoutError:
                 logger.error(
                     "Job %s(id=%s) timeout after %ss", handler_name, job_id, timeout
                 )
-                return {"code": 500, "msg": f"timeout after {timeout}s"}
+                result = {"code": 500, "msg": f"timeout after {timeout}s"}
+            except asyncio.CancelledError:
+                logger.info("Job %s(id=%s) cancelled", handler_name, job_id)
+                result = {"code": 500, "msg": "job cancelled"}
             except Exception as e:
                 logger.exception("Job %s(id=%s) failed", handler_name, job_id)
-                return {"code": 500, "msg": str(e)}
+                result = {"code": 500, "msg": str(e)}
+
+            # 回调 Admin 上报执行结果
+            await self._callback(log_id, log_date_time, result)
+            logger.info(
+                "Job %s(id=%s) callback done: %s", handler_name, job_id, result
+            )
 
-        task = asyncio.create_task(_run_with_timeout())
+        task = asyncio.create_task(_execute_and_callback())
         if job_id is not None:
             self._running_jobs[job_id] = task
 
-        try:
-            result = await task
-            return result
-        finally:
-            if job_id is not None:
-                self._running_jobs.pop(job_id, None)
+        # 添加 done 回调自动清理
+        task.add_done_callback(lambda t: self._running_jobs.pop(job_id, None))
+
+        logger.info(
+            "Job %s(id=%s) dispatched, will callback when done",
+            handler_name,
+            job_id,
+        )
+        return {"code": 200, "msg": None}
 
     async def kill(self, job_id: int) -> dict:
         """终止任务 —— Admin 调用 POST /kill"""