|
@@ -17,18 +17,6 @@ use function Opis\Closure\{serialize, unserialize};
|
|
|
*/
|
|
*/
|
|
|
abstract class BaseJob
|
|
abstract class BaseJob
|
|
|
{
|
|
{
|
|
|
- // 最大重试次数
|
|
|
|
|
- protected int $maxAttempts = 3;
|
|
|
|
|
-
|
|
|
|
|
- // 重试延迟时间(秒)
|
|
|
|
|
- protected int $retryDelay = 60;
|
|
|
|
|
-
|
|
|
|
|
- // 是否启用失败回调
|
|
|
|
|
- protected bool $enableFailedCallback = true;
|
|
|
|
|
-
|
|
|
|
|
- // 是否闭包
|
|
|
|
|
- protected bool $isClosure = false;
|
|
|
|
|
-
|
|
|
|
|
/**
|
|
/**
|
|
|
* 分发任务
|
|
* 分发任务
|
|
|
*
|
|
*
|
|
@@ -39,7 +27,7 @@ abstract class BaseJob
|
|
|
public static function dispatch(mixed $data = '', int $delay = 0, ?string $queue = null): JobDispatcher
|
|
public static function dispatch(mixed $data = '', int $delay = 0, ?string $queue = null): JobDispatcher
|
|
|
{
|
|
{
|
|
|
if ($data instanceof Closure) {
|
|
if ($data instanceof Closure) {
|
|
|
- $data = serialize($data);
|
|
|
|
|
|
|
+ $data = ['__job_closure__' => serialize($data)];
|
|
|
}
|
|
}
|
|
|
return new JobDispatcher(static::class, $data, $delay, $queue);
|
|
return new JobDispatcher(static::class, $data, $delay, $queue);
|
|
|
}
|
|
}
|
|
@@ -53,8 +41,9 @@ abstract class BaseJob
|
|
|
public function fire(Job $job, mixed $data): void
|
|
public function fire(Job $job, mixed $data): void
|
|
|
{
|
|
{
|
|
|
try {
|
|
try {
|
|
|
- if ($this->isClosure) {
|
|
|
|
|
- $data = unserialize($data);
|
|
|
|
|
|
|
+ // 闭包任务:反序列化还原
|
|
|
|
|
+ if (is_array($data) && isset($data['__job_closure__'])) {
|
|
|
|
|
+ $data = unserialize($data['__job_closure__']);
|
|
|
}
|
|
}
|
|
|
// 前置处理
|
|
// 前置处理
|
|
|
if (!$this->beforeExecute($data)) {
|
|
if (!$this->beforeExecute($data)) {
|
|
@@ -105,6 +94,8 @@ abstract class BaseJob
|
|
|
|
|
|
|
|
/**
|
|
/**
|
|
|
* 异常处理
|
|
* 异常处理
|
|
|
|
|
+ * 直接抛出异常,交 Worker 统一处理重试和失败记录。
|
|
|
|
|
+ * Worker 通过 --tries 参数控制最大重试次数,超限后自动记入 cy_failed_jobs。
|
|
|
*
|
|
*
|
|
|
* @param Job $job 队列任务对象
|
|
* @param Job $job 队列任务对象
|
|
|
* @param T $data 任务数据
|
|
* @param T $data 任务数据
|
|
@@ -117,68 +108,7 @@ abstract class BaseJob
|
|
|
'trace' => $exception->getTraceAsString()
|
|
'trace' => $exception->getTraceAsString()
|
|
|
]);
|
|
]);
|
|
|
|
|
|
|
|
- // 判断是否需要重试
|
|
|
|
|
- if ($job->attempts() < $this->maxAttempts) {
|
|
|
|
|
- // 重新发布任务
|
|
|
|
|
- $job->release($this->retryDelay);
|
|
|
|
|
- } else {
|
|
|
|
|
- // 执行失败回调
|
|
|
|
|
- if ($this->enableFailedCallback) {
|
|
|
|
|
- try {
|
|
|
|
|
- $this->onFailed($data);
|
|
|
|
|
- } catch (Exception $e) {
|
|
|
|
|
- Log::error('任务失败回调执行异常:' . $e->getMessage());
|
|
|
|
|
- }
|
|
|
|
|
- }
|
|
|
|
|
-
|
|
|
|
|
- // 重新抛出异常,让 Worker 触发 JobFailed 事件并记录到数据库
|
|
|
|
|
- throw $exception;
|
|
|
|
|
- }
|
|
|
|
|
- }
|
|
|
|
|
-
|
|
|
|
|
- /**
|
|
|
|
|
- * 任务失败处理方法 - 子类可重写
|
|
|
|
|
- *
|
|
|
|
|
- * @param T $data 任务数据
|
|
|
|
|
- */
|
|
|
|
|
- protected function onFailed(mixed $data): void
|
|
|
|
|
- {
|
|
|
|
|
- // 默认失败处理逻辑
|
|
|
|
|
- Log::error('队列任务执行失败: ' . static::class . ' - ' . json_encode($data));
|
|
|
|
|
|
|
+ throw $exception;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
- /**
|
|
|
|
|
- * 设置最大重试次数
|
|
|
|
|
- *
|
|
|
|
|
- * @param int $attempts
|
|
|
|
|
- * @return $this
|
|
|
|
|
- */
|
|
|
|
|
- protected function setMaxAttempts(int $attempts): static
|
|
|
|
|
- {
|
|
|
|
|
- $this->maxAttempts = $attempts;
|
|
|
|
|
- return $this;
|
|
|
|
|
- }
|
|
|
|
|
-
|
|
|
|
|
- /**
|
|
|
|
|
- * 设置重试延迟时间
|
|
|
|
|
- *
|
|
|
|
|
- * @param int $delay
|
|
|
|
|
- * @return $this
|
|
|
|
|
- */
|
|
|
|
|
- protected function setRetryDelay(int $delay): static
|
|
|
|
|
- {
|
|
|
|
|
- $this->retryDelay = $delay;
|
|
|
|
|
- return $this;
|
|
|
|
|
- }
|
|
|
|
|
-
|
|
|
|
|
- /**
|
|
|
|
|
- * 禁用失败回调
|
|
|
|
|
- *
|
|
|
|
|
- * @return $this
|
|
|
|
|
- */
|
|
|
|
|
- protected function disableFailedCallback(): static
|
|
|
|
|
- {
|
|
|
|
|
- $this->enableFailedCallback = false;
|
|
|
|
|
- return $this;
|
|
|
|
|
- }
|
|
|
|
|
}
|
|
}
|