494 lines
18 KiB
PHP
494 lines
18 KiB
PHP
<?php
|
||
|
||
namespace app\common\mq;
|
||
|
||
use think\Db;
|
||
use app\common\DbReconnectHelper;
|
||
use app\common\ReferenceRelevanceCheckService;
|
||
|
||
/**
|
||
* RabbitMQ 消费（队列 reference_check / ref_check.article）：
|
||
* 全局文章串行，文章内 reference_no 升序链式逐条「主题相关性」校对。
|
||
* 支持断点续跑：已完成条跳过，从卡死的 pending 条继续。
|
||
*/
|
||
class ReferenceCheckArticleWorker
|
||
{
|
||
const BATCH_WAITING = 0;
|
||
const BATCH_RUNNING = 1;
|
||
const BATCH_DONE = 2;
|
||
const BATCH_PARTIAL_FAILED = 3;
|
||
|
||
/** 批次/心跳超时：超过该秒数无 updated_at 更新视为僵尸，可抢占续跑 */
|
||
const BATCH_STALE_SECONDS = 1200;
|
||
|
||
/** @var ReferenceRelevanceCheckService */
|
||
private $svc;
|
||
|
||
public function __construct()
|
||
{
|
||
$this->svc = new ReferenceRelevanceCheckService();
|
||
}
|
||
|
||
public function handleMessage(array $payload)
|
||
{
|
||
DbReconnectHelper::ensure();
|
||
$pArticleId = intval(isset($payload['p_article_id']) ? $payload['p_article_id'] : 0);
|
||
$batchId = intval(isset($payload['batch_id']) ? $payload['batch_id'] : 0);
|
||
$trigger = isset($payload['trigger']) ? (string)$payload['trigger'] : 'enqueue';
|
||
if ($pArticleId <= 0 || $batchId <= 0) {
|
||
$this->svc->log('ReferenceCheckArticleWorker invalid payload');
|
||
return;
|
||
}
|
||
|
||
// 先释放其它文章上的僵尸 RUNNING，避免全局串行永久堵死
|
||
try {
|
||
$this->recoverStaleForeignBatches($batchId);
|
||
} catch (\Throwable $e) {
|
||
$this->svc->log('ReferenceCheckArticleWorker recoverStaleForeignBatches err=' . $e->getMessage());
|
||
}
|
||
|
||
if (!$this->canStartArticleWork($batchId)) {
|
||
$this->svc->log('ReferenceCheckArticleWorker defer batch_id=' . $batchId . ' other article running');
|
||
(new ReferenceCheckMqPublisher())->publishArticleStart(
|
||
$pArticleId,
|
||
$batchId,
|
||
isset($payload['trigger']) ? $payload['trigger'] : 'enqueue'
|
||
);
|
||
sleep(3);
|
||
return;
|
||
}
|
||
|
||
$claim = $this->claimOrResumeBatch($batchId);
|
||
if ($claim === 'skip') {
|
||
return;
|
||
}
|
||
$resume = ($claim === 'resume');
|
||
|
||
$owned = true;
|
||
$finished = false;
|
||
$attemptedCheckIds = [];
|
||
$idleRecovered = false;
|
||
try {
|
||
// 续跑时强制把卡死行收回 pending，已完成行不动
|
||
$this->svc->recoverQueueRowsForArticle($pArticleId, $resume);
|
||
if ($trigger !== 'recheck_pending_only'
|
||
&& ReferenceRelevanceCheckService::PREPARE_LITERATURE_BEFORE_CHECK) {
|
||
$this->svc->prepareLiteratureContentByArticle($pArticleId);
|
||
}
|
||
$this->svc->log(
|
||
'ReferenceCheckArticleWorker start p_article_id=' . $pArticleId
|
||
. ' batch_id=' . $batchId
|
||
. ($resume ? ' resume=1' : '')
|
||
);
|
||
|
||
while (true) {
|
||
// 组长卡在 RUNNING 时先收回，避免 worker 误跑组员报 leader not finished
|
||
$this->svc->recoverStuckRunningPendingRows($pArticleId, 180);
|
||
|
||
$row = $this->fetchNextPendingRow($pArticleId);
|
||
if (empty($row)) {
|
||
if (!$idleRecovered) {
|
||
$n = $this->svc->recoverAllRunningPendingRows($pArticleId);
|
||
if ($n > 0) {
|
||
$idleRecovered = true;
|
||
$this->svc->log(
|
||
'ReferenceCheckArticleWorker recovered ' . $n . ' running pending rows, retry fetch'
|
||
);
|
||
continue;
|
||
}
|
||
}
|
||
break;
|
||
}
|
||
$idleRecovered = false;
|
||
$checkId = $this->svc->resolveCheckRowId($row);
|
||
if ($checkId <= 0) {
|
||
continue;
|
||
}
|
||
if (!$this->svc->shouldProcessRelevanceRowNow($row)) {
|
||
if ($this->svc->recoverStuckGroupLeaderForRow($row)) {
|
||
$this->svc->log(
|
||
'ReferenceCheckArticleWorker reset stuck group leader for ref='
|
||
. intval($row['reference_no'])
|
||
);
|
||
continue;
|
||
}
|
||
$this->svc->log(
|
||
'ReferenceCheckArticleWorker skip non-leader check_id=' . $checkId
|
||
. ' ref=' . intval($row['reference_no'])
|
||
);
|
||
continue;
|
||
}
|
||
// 同一批消息内每个 check_id 只尝试一次，避免大联合组部分成功后死循环
|
||
if (isset($attemptedCheckIds[$checkId])) {
|
||
$this->svc->log('ReferenceCheckArticleWorker stop re-entry check_id=' . $checkId);
|
||
break;
|
||
}
|
||
$attemptedCheckIds[$checkId] = true;
|
||
$this->processOneRow($checkId, $row, $trigger === 'recheck_pending_only');
|
||
// 每条结束后刷新批次心跳，长文不会被误判为僵尸
|
||
$this->touchBatch($batchId);
|
||
}
|
||
|
||
$stats = $this->summarizeArticleCheckStats($pArticleId);
|
||
if (intval($stats['pending']) > 0) {
|
||
// 分块落库后仍有缺口：回 WAITING 再投递，下轮只补 pending
|
||
Db::name('article_reference_relevance_check_batch')->where('id', intval($batchId))->update([
|
||
'batch_status' => self::BATCH_WAITING,
|
||
'done_count' => intval($stats['done']),
|
||
'failed_count' => intval($stats['failed']),
|
||
'updated_at' => date('Y-m-d H:i:s'),
|
||
]);
|
||
$finished = true;
|
||
$this->svc->log(
|
||
'ReferenceCheckArticleWorker defer incomplete p_article_id=' . $pArticleId
|
||
. ' batch_id=' . $batchId
|
||
. ' pending=' . $stats['pending']
|
||
. ' done=' . $stats['done']
|
||
);
|
||
(new ReferenceCheckMqPublisher())->publishArticleStart($pArticleId, $batchId, $trigger);
|
||
} else {
|
||
$this->finalizeBatch($batchId, $stats['done'], $stats['failed'], $stats['total']);
|
||
$finished = true;
|
||
$this->svc->log(
|
||
'ReferenceCheckArticleWorker done p_article_id=' . $pArticleId
|
||
. ' batch_id=' . $batchId
|
||
. ' done=' . $stats['done']
|
||
. ' failed=' . $stats['failed']
|
||
);
|
||
$this->publishNextWaitingBatch();
|
||
}
|
||
} catch (\Throwable $e) {
|
||
// 异常不 finalize：保持 RUNNING，靠心跳超时后由后续消息断点续跑
|
||
if ($owned) {
|
||
$this->touchBatch($batchId);
|
||
$this->svc->log(
|
||
'ReferenceCheckArticleWorker abort batch_id=' . $batchId
|
||
. ' p_article_id=' . $pArticleId
|
||
. ' err=' . $e->getMessage()
|
||
);
|
||
}
|
||
throw $e;
|
||
} finally {
|
||
if ($owned && !$finished) {
|
||
// 消息进 DLQ 后主队列可能没人再推本批：主动再投递，便于稍后续跑
|
||
try {
|
||
(new ReferenceCheckMqPublisher())->publishArticleStart($pArticleId, $batchId, $trigger);
|
||
} catch (\Exception $pubErr) {
|
||
$this->svc->log('ReferenceCheckArticleWorker republish after abort failed: ' . $pubErr->getMessage());
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* @return string claim|resume|skip
|
||
*/
|
||
private function claimOrResumeBatch($batchId)
|
||
{
|
||
$batchId = intval($batchId);
|
||
$now = date('Y-m-d H:i:s');
|
||
$claimed = Db::name('article_reference_relevance_check_batch')
|
||
->where('id', $batchId)
|
||
->where('batch_status', self::BATCH_WAITING)
|
||
->update([
|
||
'batch_status' => self::BATCH_RUNNING,
|
||
'updated_at' => $now,
|
||
]);
|
||
if (intval($claimed) > 0) {
|
||
return 'claim';
|
||
}
|
||
|
||
$batch = $this->getBatch($batchId);
|
||
if (empty($batch)) {
|
||
return 'skip';
|
||
}
|
||
$status = intval($batch['batch_status']);
|
||
if ($status === self::BATCH_DONE || $status === self::BATCH_PARTIAL_FAILED) {
|
||
return 'skip';
|
||
}
|
||
if ($status === self::BATCH_RUNNING) {
|
||
if ($this->isBatchStale($batch)) {
|
||
// 僵尸 RUNNING：抢占续跑（不重置已完成明细）
|
||
Db::name('article_reference_relevance_check_batch')
|
||
->where('id', $batchId)
|
||
->where('batch_status', self::BATCH_RUNNING)
|
||
->update(['updated_at' => $now]);
|
||
$this->svc->log('ReferenceCheckArticleWorker reclaim stale batch_id=' . $batchId);
|
||
return 'resume';
|
||
}
|
||
// 仍有心跳，说明别的消费者在跑本批
|
||
return 'skip';
|
||
}
|
||
return 'skip';
|
||
}
|
||
|
||
private function isBatchStale(array $batch)
|
||
{
|
||
$updatedAt = isset($batch['updated_at']) ? strtotime((string)$batch['updated_at']) : 0;
|
||
if ($updatedAt <= 0) {
|
||
return true;
|
||
}
|
||
return (time() - $updatedAt) >= self::BATCH_STALE_SECONDS;
|
||
}
|
||
|
||
private function touchBatch($batchId)
|
||
{
|
||
Db::name('article_reference_relevance_check_batch')
|
||
->where('id', intval($batchId))
|
||
->where('batch_status', self::BATCH_RUNNING)
|
||
->update(['updated_at' => date('Y-m-d H:i:s')]);
|
||
}
|
||
|
||
/**
|
||
* 其它文章僵尸 RUNNING → 改回 WAITING 并重新入队，从卡死条续跑
|
||
*/
|
||
private function recoverStaleForeignBatches($exceptBatchId)
|
||
{
|
||
$exceptBatchId = intval($exceptBatchId);
|
||
$staleBefore = date('Y-m-d H:i:s', time() - self::BATCH_STALE_SECONDS);
|
||
// ThinkPHP 5：whereRaw 绑参会与命名占位符冲突(HY093)，日期直接拼进 SQL
|
||
$staleSql = "(updated_at IS NULL OR updated_at < '" . addslashes($staleBefore) . "')";
|
||
$rows = Db::name('article_reference_relevance_check_batch')
|
||
->where('batch_status', self::BATCH_RUNNING)
|
||
->where('id', '<>', $exceptBatchId)
|
||
->whereRaw($staleSql)
|
||
->order('id asc')
|
||
->limit(20)
|
||
->select();
|
||
if (empty($rows)) {
|
||
return;
|
||
}
|
||
|
||
$publisher = new ReferenceCheckMqPublisher();
|
||
foreach ($rows as $row) {
|
||
$bid = intval($row['id']);
|
||
$pid = intval($row['p_article_id']);
|
||
$affected = Db::name('article_reference_relevance_check_batch')
|
||
->where('id', $bid)
|
||
->where('batch_status', self::BATCH_RUNNING)
|
||
->whereRaw($staleSql)
|
||
->update([
|
||
'batch_status' => self::BATCH_WAITING,
|
||
'updated_at' => date('Y-m-d H:i:s'),
|
||
]);
|
||
if (intval($affected) <= 0) {
|
||
continue;
|
||
}
|
||
// 强制收回卡死行；已完成条保持不动
|
||
$this->svc->recoverQueueRowsForArticle($pid, true);
|
||
$this->svc->log('ReferenceCheckArticleWorker recover stale foreign batch_id=' . $bid . ' p_article_id=' . $pid);
|
||
try {
|
||
$publisher->publishArticleStart(
|
||
$pid,
|
||
$bid,
|
||
isset($row['trigger']) ? $row['trigger'] : 'enqueue'
|
||
);
|
||
} catch (\Exception $e) {
|
||
$this->svc->log('ReferenceCheckArticleWorker recover publish failed batch_id=' . $bid . ' err=' . $e->getMessage());
|
||
}
|
||
}
|
||
}
|
||
|
||
private function summarizeArticleCheckStats($pArticleId)
|
||
{
|
||
$rows = Db::name('article_reference_relevance_check_result')
|
||
->where('p_article_id', intval($pArticleId))
|
||
->field('status')
|
||
->select();
|
||
$done = 0;
|
||
$failed = 0;
|
||
$pending = 0;
|
||
foreach ($rows as $row) {
|
||
$st = intval(isset($row['status']) ? $row['status'] : -1);
|
||
if ($st === ReferenceRelevanceCheckService::RECORD_COMPLETED) {
|
||
$done++;
|
||
} elseif ($st === ReferenceRelevanceCheckService::RECORD_FAILED) {
|
||
$failed++;
|
||
} elseif ($st === ReferenceRelevanceCheckService::RECORD_PENDING) {
|
||
$pending++;
|
||
}
|
||
}
|
||
return [
|
||
'done' => $done,
|
||
'failed' => $failed,
|
||
'pending' => $pending,
|
||
'total' => $done + $failed + $pending,
|
||
];
|
||
}
|
||
|
||
private function canStartArticleWork($batchId)
|
||
{
|
||
$running = Db::name('article_reference_relevance_check_batch')
|
||
->where('batch_status', self::BATCH_RUNNING)
|
||
->where('id', '<>', intval($batchId))
|
||
->count();
|
||
return intval($running) === 0;
|
||
}
|
||
|
||
private function getBatch($batchId)
|
||
{
|
||
return Db::name('article_reference_relevance_check_batch')->where('id', intval($batchId))->find();
|
||
}
|
||
|
||
private function fetchNextPendingRow($pArticleId)
|
||
{
|
||
$rows = Db::name('article_reference_relevance_check_result')
|
||
->where('p_article_id', intval($pArticleId))
|
||
->where('queue_status', ReferenceRelevanceCheckService::QUEUE_PENDING)
|
||
->where('status', ReferenceRelevanceCheckService::RECORD_PENDING)
|
||
->order('reference_no asc,am_id asc,text_start asc,id asc')
|
||
->limit(100)
|
||
->select();
|
||
if (empty($rows)) {
|
||
return null;
|
||
}
|
||
foreach ($rows as $row) {
|
||
if ($this->svc->shouldProcessRelevanceRowNow($row)) {
|
||
return $row;
|
||
}
|
||
}
|
||
return null;
|
||
}
|
||
|
||
/**
|
||
* @return string ok|failed|skip
|
||
*/
|
||
private function processOneRow($checkId, array $row, $skipLiteratureFetch = false)
|
||
{
|
||
DbReconnectHelper::ensure();
|
||
$claimed = Db::name('article_reference_relevance_check_result')
|
||
->where('id', intval($checkId))
|
||
->where('queue_status', ReferenceRelevanceCheckService::QUEUE_PENDING)
|
||
->update([
|
||
'queue_status' => ReferenceRelevanceCheckService::QUEUE_RUNNING,
|
||
'updated_at' => date('Y-m-d H:i:s'),
|
||
]);
|
||
if (intval($claimed) <= 0) {
|
||
return 'skip';
|
||
}
|
||
|
||
$retryCount = intval(isset($row['retry_count']) ? $row['retry_count'] : 0);
|
||
try {
|
||
$this->svc->runCheckOnce($checkId, $skipLiteratureFetch);
|
||
DbReconnectHelper::ensure();
|
||
$fresh = Db::name('article_reference_relevance_check_result')->where('id', intval($checkId))->find();
|
||
if (empty($fresh)) {
|
||
return 'skip';
|
||
}
|
||
$st = intval($fresh['status']);
|
||
if ($st === ReferenceRelevanceCheckService::RECORD_COMPLETED) {
|
||
$this->svc->markQueueRuntime($checkId, ReferenceRelevanceCheckService::QUEUE_COMPLETED, $retryCount);
|
||
return 'ok';
|
||
}
|
||
if ($st === ReferenceRelevanceCheckService::RECORD_FAILED) {
|
||
$this->svc->markQueueRuntime($checkId, ReferenceRelevanceCheckService::QUEUE_FAILED, $retryCount);
|
||
return 'failed';
|
||
}
|
||
// 分块部分成功：本行仍 pending，留给后续消息补跑
|
||
$this->svc->markQueueRuntime($checkId, ReferenceRelevanceCheckService::QUEUE_PENDING, $retryCount);
|
||
return 'ok';
|
||
} catch (\Exception $e) {
|
||
$this->svc->log('ReferenceCheckArticleWorker check_id=' . $checkId . ' err=' . $e->getMessage());
|
||
DbReconnectHelper::ensure();
|
||
// 联合组组员被提前领取：不算失败，交还 pending 等组长
|
||
if (intval($e->getCode()) === 9001
|
||
|| strpos($e->getMessage(), 'Citation group leader not finished') !== false) {
|
||
$this->svc->markQueueRuntime($checkId, ReferenceRelevanceCheckService::QUEUE_PENDING, $retryCount);
|
||
return 'skip';
|
||
}
|
||
if ($this->svc->isRelevanceLlmFailureMessage($e->getMessage())) {
|
||
$this->svc->markQueueRuntime($checkId, ReferenceRelevanceCheckService::QUEUE_PENDING, $retryCount);
|
||
return 'ok';
|
||
}
|
||
try {
|
||
$fresh = Db::name('article_reference_relevance_check_result')->where('id', intval($checkId))->find();
|
||
if (!empty($fresh) && intval($fresh['status']) === ReferenceRelevanceCheckService::RECORD_COMPLETED) {
|
||
// 异常前已有分块落库成功，保留成果，本行按完成处理
|
||
$this->svc->markQueueRuntime($checkId, ReferenceRelevanceCheckService::QUEUE_COMPLETED, $retryCount);
|
||
return 'ok';
|
||
}
|
||
if (!empty($fresh) && intval($fresh['status']) === ReferenceRelevanceCheckService::RECORD_FAILED) {
|
||
if (intval($fresh['queue_status']) !== ReferenceRelevanceCheckService::QUEUE_FAILED) {
|
||
$this->svc->markQueueRuntime($checkId, ReferenceRelevanceCheckService::QUEUE_FAILED, $retryCount);
|
||
}
|
||
return 'failed';
|
||
}
|
||
$groupRows = !empty($fresh) ? $this->svc->findCitationGroupRowsForWorker($fresh) : [];
|
||
if (!empty($groupRows)) {
|
||
// 只失败仍未完成的行，已 completed 的分块结果保留
|
||
$incomplete = [];
|
||
foreach ($groupRows as $gr) {
|
||
if (intval($gr['status']) !== ReferenceRelevanceCheckService::RECORD_COMPLETED) {
|
||
$incomplete[] = $gr;
|
||
}
|
||
}
|
||
if (!empty($incomplete)) {
|
||
$this->svc->failGroupWithQueue($incomplete, $e->getMessage(), $retryCount);
|
||
}
|
||
} else {
|
||
$this->svc->updateCheckResult($checkId, [
|
||
'status' => ReferenceRelevanceCheckService::RECORD_FAILED,
|
||
'error_msg' => $e->getMessage(),
|
||
]);
|
||
$this->svc->markQueueRuntime($checkId, ReferenceRelevanceCheckService::QUEUE_FAILED, $retryCount);
|
||
}
|
||
} catch (\Exception $e2) {
|
||
\think\Log::error('ReferenceCheckArticleWorker markFailed: ' . $e2->getMessage());
|
||
}
|
||
return 'failed';
|
||
}
|
||
}
|
||
|
||
private function finalizeBatch($batchId, $done, $failed, $total = 0)
|
||
{
|
||
$batch = $this->getBatch($batchId);
|
||
if (empty($batch)) {
|
||
return;
|
||
}
|
||
$done = intval($done);
|
||
$failed = intval($failed);
|
||
$total = intval($total);
|
||
if ($total <= 0) {
|
||
$total = intval($batch['total_count']);
|
||
}
|
||
if (($done + $failed) > $total) {
|
||
$total = $done + $failed;
|
||
}
|
||
$status = self::BATCH_DONE;
|
||
if ($failed > 0) {
|
||
$status = self::BATCH_PARTIAL_FAILED;
|
||
}
|
||
Db::name('article_reference_relevance_check_batch')->where('id', intval($batchId))->update([
|
||
'batch_status' => $status,
|
||
'total_count' => $total,
|
||
'done_count' => $done,
|
||
'failed_count' => $failed,
|
||
'updated_at' => date('Y-m-d H:i:s'),
|
||
]);
|
||
if ($total > 0 && ($done + $failed) < $total) {
|
||
$this->svc->log('ReferenceCheckArticleWorker batch_id=' . $batchId . ' incomplete total=' . $total . ' done=' . $done . ' failed=' . $failed);
|
||
}
|
||
}
|
||
|
||
private function publishNextWaitingBatch()
|
||
{
|
||
$next = Db::name('article_reference_relevance_check_batch')
|
||
->where('batch_status', self::BATCH_WAITING)
|
||
->order('id asc')
|
||
->find();
|
||
if (empty($next)) {
|
||
return;
|
||
}
|
||
try {
|
||
(new ReferenceCheckMqPublisher())->publishArticleStart(
|
||
intval($next['p_article_id']),
|
||
intval($next['id']),
|
||
isset($next['trigger']) ? $next['trigger'] : 'enqueue'
|
||
);
|
||
} catch (\Exception $e) {
|
||
$this->svc->log('ReferenceCheck publishNextWaitingBatch failed: ' . $e->getMessage());
|
||
\think\Log::error('ReferenceCheck publishNextWaitingBatch: ' . $e->getMessage());
|
||
}
|
||
}
|
||
}
|