zm/tougao
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

一处原文引用20多个参考文献优化

Browse Source
This commit is contained in:
wyn
2026-08-04 17:11:53 +08:00
parent 485d7c8d97
commit 32fb903c5d
4 changed files with 559 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files

287
application/common/mq/ReferenceCheckArticleWorker.php
View File

@@ -9,6 +9,7 @@ use app\common\ReferenceRelevanceCheckService;
/**
* RabbitMQ 消费（队列 reference_check / ref_check.article）：
* 全局文章串行，文章内 reference_no 升序链式逐条「主题相关性」校对。
* 支持断点续跑：已完成条跳过，从卡死的 pending 条继续。
*/
class ReferenceCheckArticleWorker
{
@@ -17,6 +18,9 @@ class ReferenceCheckArticleWorker
const BATCH_DONE = 2;
const BATCH_PARTIAL_FAILED = 3;
/** 批次/心跳超时：超过该秒数无 updated_at 更新视为僵尸，可抢占续跑 */
const BATCH_STALE_SECONDS = 1200;
/** @var ReferenceRelevanceCheckService */
private $svc;
@@ -36,6 +40,9 @@ class ReferenceCheckArticleWorker
return;
}
// 先释放其它文章上的僵尸 RUNNING，避免全局串行永久堵死
$this->recoverStaleForeignBatches($batchId);
if (!$this->canStartArticleWork($batchId)) {
$this->svc->log('ReferenceCheckArticleWorker defer batch_id=' . $batchId . ' other article running');
(new ReferenceCheckMqPublisher())->publishArticleStart(
@@ -47,95 +54,206 @@ class ReferenceCheckArticleWorker
return;
}
if (!$this->claimBatch($batchId)) {
$batch = $this->getBatch($batchId);
// 已被其他消费者领取或已结束，当前消息直接跳过，避免同批次并发重复执行
if (empty($batch)
|| intval($batch['batch_status']) === self::BATCH_RUNNING
|| intval($batch['batch_status']) === self::BATCH_DONE
|| intval($batch['batch_status']) === self::BATCH_PARTIAL_FAILED) {
return;
$claim = $this->claimOrResumeBatch($batchId);
if ($claim === 'skip') {
return;
}
$resume = ($claim === 'resume');
$owned = true;
$finished = 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) {
$row = $this->fetchNextPendingRow($pArticleId);
if (empty($row)) {
break;
}
$checkId = $this->svc->resolveCheckRowId($row);
if ($checkId <= 0) {
continue;
}
$this->processOneRow($checkId, $row, $trigger === 'recheck_pending_only');
// 每条结束后刷新批次心跳，长文不会被误判为僵尸
$this->touchBatch($batchId);
}
$stats = $this->summarizeArticleCheckStats($pArticleId);
$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());
}
}
}
$this->svc->recoverQueueRowsForArticle($pArticleId);
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);
// 快照本批待处理 id：联合引用组长一次会整组落库，循环计数会小于 total_count，收尾按快照回填
$trackedIds = $this->listPendingCheckIds($pArticleId);
$done = 0;
$failed = 0;
while (true) {
$row = $this->fetchNextPendingRow($pArticleId);
if (empty($row)) {
break;
}
$checkId = $this->svc->resolveCheckRowId($row);
if ($checkId <= 0) {
continue;
}
$result = $this->processOneRow($checkId, $row, $trigger === 'recheck_pending_only');
if ($result === 'ok') {
$done++;
} elseif ($result === 'failed') {
$failed++;
}
}
if (!empty($trackedIds)) {
$stats = $this->summarizeTrackedCheckIds($trackedIds);
$done = intval($stats['done']);
$failed = intval($stats['failed']);
}
$this->finalizeBatch($batchId, $done, $failed);
$this->svc->log('ReferenceCheckArticleWorker done p_article_id=' . $pArticleId . ' batch_id=' . $batchId . ' done=' . $done . ' failed=' . $failed);
$this->publishNextWaitingBatch();
}
private function listPendingCheckIds($pArticleId)
/**
* @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);
$rows = Db::name('article_reference_relevance_check_batch')
->where('batch_status', self::BATCH_RUNNING)
->where('id', '<>', $exceptBatchId)
->whereRaw('(updated_at IS NULL OR updated_at < ?)', [$staleBefore])
->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('(updated_at IS NULL OR updated_at < ?)', [$staleBefore])
->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))
->where('queue_status', ReferenceRelevanceCheckService::QUEUE_PENDING)
->where('status', ReferenceRelevanceCheckService::RECORD_PENDING)
->field('id')
->select();
$ids = [];
foreach ($rows as $row) {
$id = intval(isset($row['id']) ? $row['id'] : 0);
if ($id > 0) {
$ids[] = $id;
}
}
return $ids;
}
private function summarizeTrackedCheckIds(array $checkIds)
{
$checkIds = array_values(array_filter(array_map('intval', $checkIds)));
if (empty($checkIds)) {
return ['done' => 0, 'failed' => 0];
}
$rows = Db::name('article_reference_relevance_check_result')
->whereIn('id', $checkIds)
->field('id,status')
->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];
return [
'done' => $done,
'failed' => $failed,
'pending' => $pending,
'total' => $done + $failed + $pending,
];
}
private function canStartArticleWork($batchId)
@@ -147,20 +265,6 @@ class ReferenceCheckArticleWorker
return intval($running) === 0;
}
private function claimBatch($batchId)
{
$now = date('Y-m-d H:i:s');
$affected = Db::name('article_reference_relevance_check_batch')
->where('id', intval($batchId))
// 只允许 WAITING -> RUNNING，禁止已 RUNNING 的批次被重复 claim
->where('batch_status', self::BATCH_WAITING)
->update([
'batch_status' => self::BATCH_RUNNING,
'updated_at' => $now,
]);
return intval($affected) > 0;
}
private function getBatch($batchId)
{
return Db::name('article_reference_relevance_check_batch')->where('id', intval($batchId))->find();
@@ -185,7 +289,10 @@ class ReferenceCheckArticleWorker
$claimed = Db::name('article_reference_relevance_check_result')
->where('id', intval($checkId))
->where('queue_status', ReferenceRelevanceCheckService::QUEUE_PENDING)
->update(['queue_status' => ReferenceRelevanceCheckService::QUEUE_RUNNING]);
->update([
'queue_status' => ReferenceRelevanceCheckService::QUEUE_RUNNING,
'updated_at' => date('Y-m-d H:i:s'),
]);
if (intval($claimed) <= 0) {
return 'skip';
}
@@ -223,16 +330,18 @@ class ReferenceCheckArticleWorker
}
}
private function finalizeBatch($batchId, $done, $failed)
private function finalizeBatch($batchId, $done, $failed, $total = 0)
{
$batch = $this->getBatch($batchId);
if (empty($batch)) {
return;
}
$total = intval($batch['total_count']);
$done = intval($done);
$failed = intval($failed);
// 快照回填后若实际终态条数多于入队 total，抬升 total 保持一致
$total = intval($total);
if ($total <= 0) {
$total = intval($batch['total_count']);
}
if (($done + $failed) > $total) {
$total = $done + $failed;
}