mirror of
https://gitee.com/ulthon/ulthon_admin.git
synced 2026-11-08 03:50:50 +08:00
223 lines
8.4 KiB
PHP
223 lines
8.4 KiB
PHP
<?php
|
||
|
||
declare(strict_types=1);
|
||
|
||
namespace base\tools\controller\timer;
|
||
|
||
use app\common\controller\TimerController;
|
||
use app\common\service\NoticeService;
|
||
use base\admin\model\SystemNoticeTargetBase;
|
||
use think\facade\Db;
|
||
use think\facade\Log;
|
||
|
||
/**
|
||
* 通知异步投递消费者（Base 层）.
|
||
*
|
||
* 消费 system_notice_target 中 status=0（待发送）的 email/sms/wechat 渠道行：
|
||
* inmail 行由 NoticeService 直接写成功终态永不入队，故候选查询无需渠道过滤。
|
||
*
|
||
* 消费模式（ulthon-queue 模式 A：DB 原子抢占）：
|
||
* 1. 候选查询走 (status, id) 复合索引 idx_status_id，只读前 BATCH_SIZE 条
|
||
* 2. 原子抢占：读和标记压进一条条件 UPDATE（WHERE 带 status=0），
|
||
* concurrency>1 时被抢先的行 affected=0，竞态在此消解；不按
|
||
* concurrencyId 划分数据（反模式：id % N 取余分片导致索引失效）
|
||
* 3. 逐条投递：调 NoticeService::channels() 对应驱动 send(target, notice)，
|
||
* 单条失败不影响同批其他行
|
||
* 4. 回写幂等：WHERE 含 claim_token + status=1，防回写他人抢占行；
|
||
* 失败走指数退避重试（不删失败任务），第 MAX_ATTEMPTS 次尝试失败置终态
|
||
*
|
||
* 时间窗口模式（ulthon-timer）：do() 内 maxRunTime 窗口循环 doBatch()，
|
||
* 超时自动退出由定时器下次调度继续；frequency >= maxRunTime * 2。
|
||
*
|
||
* 调度与防刷：$frequency = null 显式关闭控制器侧防刷（手动触发不受
|
||
* "请不要频繁请求"拦截），调度节流完全由定时器侧 Cache 节流承担。
|
||
*
|
||
* 依赖倒置：渠道映射经 app\common\service\NoticeService 入口获取
|
||
* （双层双向注入惯例：base 层引用 app 入口类）；业务侧如需定制消费
|
||
* 行为，重写 app/tools/controller/timer/NoticeConsumer.php 对应方法即可。
|
||
*/
|
||
class NoticeConsumerBase extends TimerController
|
||
{
|
||
/**
|
||
* 防刷间隔：null = 关闭控制器侧防刷（调度由定时器侧 frequency 节流）.
|
||
*/
|
||
protected $frequency = null;
|
||
|
||
/**
|
||
* 并发上限 cap：2 分片（id ∈ [0, 1]），竞争消费由原子抢占消解.
|
||
*/
|
||
protected $concurrency = 2;
|
||
|
||
/**
|
||
* 时间窗口（秒）：3 配 frequency=10s，满足 frequency >= maxRunTime * 2.
|
||
*/
|
||
protected $maxRunTime = 3;
|
||
|
||
/**
|
||
* 单批候选数：走 (status, id) 索引区间扫描，与表总量无关.
|
||
*/
|
||
protected const BATCH_SIZE = 20;
|
||
|
||
/**
|
||
* 最大尝试次数：首次 + 4 次重试 = 共 5 次，第 5 次失败置 STATUS_FAILED 终态.
|
||
*/
|
||
protected const MAX_ATTEMPTS = 5;
|
||
|
||
/**
|
||
* 失败退避基准（秒）：next_retry_time = now + 基准 * 2^retry_count
|
||
* （retry_count 为自增后的值），最大 60 * 2^4 = 960 秒.
|
||
*/
|
||
protected const RETRY_BASE_SECONDS = 60;
|
||
|
||
public function do()
|
||
{
|
||
$max_time = time() + $this->maxRunTime;
|
||
|
||
do {
|
||
$has_work = $this->doBatch();
|
||
|
||
if (!$has_work) {
|
||
sleep(1); // 队列空时休眠，防空转；有数据时不要 sleep
|
||
break;
|
||
}
|
||
|
||
if (time() >= $max_time) {
|
||
break;
|
||
}
|
||
} while (true);
|
||
|
||
return 'done';
|
||
}
|
||
|
||
/**
|
||
* 消费一批候选（抢占 + 逐条投递）.
|
||
*
|
||
* @return bool 是否有工作量（无候选返回 false，do() 据此休眠退出）
|
||
*/
|
||
protected function doBatch(): bool
|
||
{
|
||
// 候选：status=0 且退避时间已到，走 (status, id) 索引只读前 20 条
|
||
$candidates = Db::name('system_notice_target')
|
||
->where('status', SystemNoticeTargetBase::STATUS_PENDING)
|
||
->where('next_retry_time', '<=', time())
|
||
->order('id')
|
||
->limit(self::BATCH_SIZE)
|
||
->column('id');
|
||
|
||
if (empty($candidates)) {
|
||
return false;
|
||
}
|
||
|
||
// 原子抢占：WHERE 带 status=0，被其他消费者抢先的行 affected=0，竞态在此消解
|
||
$token = md5(uniqid('', true) . '-' . mt_rand());
|
||
Db::name('system_notice_target')
|
||
->where('id', 'in', $candidates)
|
||
->where('status', SystemNoticeTargetBase::STATUS_PENDING)
|
||
->update([
|
||
'status' => SystemNoticeTargetBase::STATUS_SENDING,
|
||
'claim_token' => $token,
|
||
'claim_time' => time(),
|
||
'update_time' => time(),
|
||
]);
|
||
|
||
// 查回真正抢到的：必须带 id IN（主键定位），只按 claim_token 查会全表扫描
|
||
$mine = Db::name('system_notice_target')
|
||
->where('id', 'in', $candidates)
|
||
->where('claim_token', $token)
|
||
->select()
|
||
->toArray();
|
||
|
||
if (empty($mine)) {
|
||
return false;
|
||
}
|
||
|
||
// 批量取通知主体（title/content 驱动报文所需；notice 表无 claim 概念）
|
||
$notices = Db::name('system_notice')
|
||
->whereIn('id', array_values(array_unique(array_column($mine, 'notice_id'))))
|
||
->select()
|
||
->toArray();
|
||
$noticeMap = array_column($notices, null, 'id');
|
||
|
||
$channels = NoticeService::channels();
|
||
|
||
foreach ($mine as $item) {
|
||
try {
|
||
$driverClass = $channels[$item['channel']] ?? null;
|
||
if ($driverClass === null) {
|
||
throw new \RuntimeException('未注册的投递渠道: ' . $item['channel']);
|
||
}
|
||
|
||
$notice = $noticeMap[$item['notice_id']] ?? null;
|
||
if ($notice === null) {
|
||
throw new \RuntimeException('通知主体不存在: notice_id=' . $item['notice_id']);
|
||
}
|
||
|
||
$driver = new $driverClass();
|
||
$driver->send($item, $notice);
|
||
|
||
// 成功幂等回写：WHERE 含 claim_token（防回写他人抢占行）+ status=1 兜底
|
||
Db::name('system_notice_target')
|
||
->where('id', $item['id'])
|
||
->where('claim_token', $token)
|
||
->where('status', SystemNoticeTargetBase::STATUS_SENDING)
|
||
->update([
|
||
'status' => SystemNoticeTargetBase::STATUS_SUCCESS,
|
||
'send_time' => time(),
|
||
'update_time' => time(),
|
||
]);
|
||
} catch (\Throwable $e) {
|
||
$this->handleFailure($item, $token, $e);
|
||
}
|
||
}
|
||
|
||
return true;
|
||
}
|
||
|
||
/**
|
||
* 单条失败处理：指数退避重试，第 MAX_ATTEMPTS 次尝试失败置终态（不删失败任务）.
|
||
*
|
||
* 退避公式：next_retry_time = now + 60 * 2^retry_count（retry_count 为自增后值）；
|
||
* 回写 WHERE 含 claim_token + status=1，幂等兜底重复消费。
|
||
*
|
||
* @param array $item 抢占到的投递行
|
||
* @param string $token 本轮抢占令牌
|
||
* @param \Throwable $e 投递异常
|
||
*/
|
||
protected function handleFailure(array $item, string $token, \Throwable $e): void
|
||
{
|
||
$retry_count = (int) $item['retry_count'] + 1;
|
||
$error_message = mb_substr($e->getMessage(), 0, 500);
|
||
|
||
Log::error('notice consumer: 投递失败 target_id=' . $item['id']
|
||
. ' channel=' . $item['channel'] . ' retry=' . $retry_count . ' - ' . $error_message);
|
||
|
||
if ($retry_count >= self::MAX_ATTEMPTS) {
|
||
// 第 5 次尝试失败：终态
|
||
Db::name('system_notice_target')
|
||
->where('id', $item['id'])
|
||
->where('claim_token', $token)
|
||
->where('status', SystemNoticeTargetBase::STATUS_SENDING)
|
||
->update([
|
||
'status' => SystemNoticeTargetBase::STATUS_FAILED,
|
||
'retry_count' => $retry_count,
|
||
'error_message' => $error_message,
|
||
'update_time' => time(),
|
||
]);
|
||
|
||
return;
|
||
}
|
||
|
||
Db::name('system_notice_target')
|
||
->where('id', $item['id'])
|
||
->where('claim_token', $token)
|
||
->where('status', SystemNoticeTargetBase::STATUS_SENDING)
|
||
->update([
|
||
'status' => SystemNoticeTargetBase::STATUS_PENDING,
|
||
'retry_count' => $retry_count,
|
||
'next_retry_time' => time() + self::RETRY_BASE_SECONDS * (2 ** $retry_count),
|
||
'error_message' => $error_message,
|
||
'update_time' => time(),
|
||
]);
|
||
}
|
||
}
|