71 lines
2.6 KiB
PHP
71 lines
2.6 KiB
PHP
<?php
|
||
namespace app\api\job;
|
||
|
||
use think\queue\Job;
|
||
use app\common\QueueJob;
|
||
use app\common\QueueRedis;
|
||
use app\common\ReferenceDispatchService;
|
||
|
||
/**
|
||
* 单条参考文献分流（journal/book/other）异步任务。
|
||
* freshRefers 按 p_refer_id 逐条入队，多条文献由多个 worker 并行处理。
|
||
*
|
||
* Worker：
|
||
* php think queue:work --queue ReferenceDispatchQueue,ArticleReferDetailQueue
|
||
*/
|
||
class ReferenceDispatchQueue
|
||
{
|
||
private $oQueueJob;
|
||
private $QueueRedis;
|
||
private $completedExprie = 3600;
|
||
|
||
public function __construct()
|
||
{
|
||
$this->oQueueJob = new QueueJob;
|
||
$this->QueueRedis = QueueRedis::getInstance();
|
||
}
|
||
|
||
public function fire(Job $job, $data)
|
||
{
|
||
$this->oQueueJob->init($job);
|
||
|
||
$rawBody = empty($job->getRawBody()) ? '' : $job->getRawBody();
|
||
$jobData = empty($rawBody) ? [] : json_decode($rawBody, true);
|
||
$jobId = empty($jobData['id']) ? 'unknown' : $jobData['id'];
|
||
|
||
$this->oQueueJob->log("-----------队列任务开始-----------");
|
||
$this->oQueueJob->log("当前任务ID: {$jobId}, 尝试次数: {$job->attempts()}");
|
||
|
||
$iPArticleId = empty($data['p_article_id']) ? 0 : intval($data['p_article_id']);
|
||
$iPReferId = empty($data['p_refer_id']) ? 0 : intval($data['p_refer_id']);
|
||
if (empty($iPArticleId) || empty($iPReferId)) {
|
||
$this->oQueueJob->log("无效的 p_article_id 或 p_refer_id，删除任务");
|
||
$job->delete();
|
||
return;
|
||
}
|
||
|
||
try {
|
||
$sClassName = get_class($this);
|
||
$sRedisKey = "queue_job:{$sClassName}:{$iPArticleId}:{$iPReferId}";
|
||
$sRedisValue = uniqid() . '_' . getmypid();
|
||
if (!$this->oQueueJob->acquireLock($sRedisKey, $sRedisValue, $job)) {
|
||
return;
|
||
}
|
||
|
||
(new ReferenceDispatchService())->dispatchReferByType($iPReferId);
|
||
|
||
$this->QueueRedis->finishJob($sRedisKey, 'completed', $this->completedExprie, $sRedisValue);
|
||
$job->delete();
|
||
$this->oQueueJob->log("任务执行成功 | 日志ID: {$sRedisKey} | p_refer_id:{$iPReferId}");
|
||
} catch (\RuntimeException $e) {
|
||
$this->oQueueJob->handleRetryableException($e, $sRedisKey, $sRedisValue, $job);
|
||
} catch (\LogicException $e) {
|
||
$this->oQueueJob->handleNonRetryableException($e, $sRedisKey, $sRedisValue, $job);
|
||
} catch (\Exception $e) {
|
||
$this->oQueueJob->handleRetryableException($e, $sRedisKey, $sRedisValue, $job);
|
||
} finally {
|
||
$this->oQueueJob->finnal();
|
||
}
|
||
}
|
||
}
|