|
|
@@ -138,7 +138,7 @@ class XxlJobExecutor:
|
|
|
except Exception:
|
|
|
pass
|
|
|
|
|
|
- async def _callback(self, log_id: int, log_date_time: int, result: dict) -> None:
|
|
|
+ async def _callback(self, log_id: int, log_date_time: int, result: dict) -> bool:
|
|
|
"""向 Admin 回调上报任务执行结果 —— POST /api/callback
|
|
|
|
|
|
对标 Java 版的 TriggerCallbackThread.callback()。
|
|
|
@@ -147,10 +147,13 @@ class XxlJobExecutor:
|
|
|
log_id: XXL-JOB 调度日志 ID
|
|
|
log_date_time: 调度时间戳(毫秒)
|
|
|
result: handler 返回的 {"code": 200, "msg": "..."} 或异常结果
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ True 如果至少向一个 Admin 成功上报,False 全部失败
|
|
|
"""
|
|
|
if not log_id:
|
|
|
logger.warning("Skip callback: logId is empty")
|
|
|
- return
|
|
|
+ return False
|
|
|
|
|
|
body = {
|
|
|
"logId": log_id,
|
|
|
@@ -168,13 +171,14 @@ class XxlJobExecutor:
|
|
|
if resp.status == 200:
|
|
|
data = await resp.json()
|
|
|
if data.get("code") == 200:
|
|
|
- return
|
|
|
+ return True
|
|
|
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)
|
|
|
+ return False
|
|
|
|
|
|
async def _heartbeat_loop(self) -> None:
|
|
|
"""定期心跳:每隔 registry_interval 秒重新注册(续约)"""
|
|
|
@@ -260,8 +264,14 @@ class XxlJobExecutor:
|
|
|
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)
|
|
|
+ callback_ok = await self._callback(log_id, log_date_time, result)
|
|
|
+ if callback_ok:
|
|
|
+ logger.info(
|
|
|
+ "Job %s(id=%s) callback succeeded: %s",
|
|
|
+ handler_name,
|
|
|
+ job_id,
|
|
|
+ result,
|
|
|
+ )
|
|
|
|
|
|
task = asyncio.create_task(_execute_and_callback())
|
|
|
if job_id is not None:
|