feat(notice): timer 异步投递消费者与超时回收任务

2026-11-07 19:40:49 +08:00 · 2026-10-12 12:41:03 +00:00
parent d4004f2f1b
commit ca16cd56de
5 changed files with 314 additions and 0 deletions
--- a/extend/base/tools/controller/timer/NoticeConsumerBase.php
+++ b/extend/base/tools/controller/timer/NoticeConsumerBase.php
@@ -0,0 +1,222 @@
+<?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(),
+            ]);
+    }
+}
--- a/extend/base/tools/controller/timer/NoticeReclaimBase.php
+++ b/extend/base/tools/controller/timer/NoticeReclaimBase.php
@@ -0,0 +1,56 @@
+<?php
+
+declare(strict_types=1);
+
+namespace base\tools\controller\timer;
+
+use app\common\controller\TimerController;
+use base\admin\model\SystemNoticeTargetBase;
+use think\facade\Db;
+
+/**
+ * 通知投递超时回收任务（Base 层）.
+ *
+ * 消费者崩溃后 status=1（发送中）的行会永久卡住，本任务按抢占时间
+ * 超时回收重置回待发送，交由消费者重新抢占（at-least-once 语义，
+ * 渠道驱动幂等性由发送侧业务兜底）。
+ *
+ * 回收阈值必须大于消费者单条最大处理时长，否则会误回收正在处理的行
+ * （超时重投的重复消费比永久卡行好处理：回写 WHERE 含 claim_token，
+ * 旧令牌回写自动失效）。
+ *
+ * 回收本身是原子条件 UPDATE，幂等无害；多节点部署 run_type=auto
+ * 单节点跑即可（回收与消费者无节点亲和）。
+ */
+class NoticeReclaimBase extends TimerController
+{
+    /**
+     * 防刷间隔（秒）：回收是低频兜底任务，防外部重复请求.
+     */
+    protected $frequency = 60;
+
+    /**
+     * 并发上限 cap：单分片（id=0），回收全局单份执行.
+     */
+    protected $concurrency = 1;
+
+    /**
+     * 回收阈值（秒）：必须大于消费者单条最大处理时长.
+     */
+    protected const RECLAIM_TIMEOUT = 600;
+
+    public function do()
+    {
+        $affected = Db::name('system_notice_target')
+            ->where('status', SystemNoticeTargetBase::STATUS_SENDING)
+            ->where('claim_time', '<', time() - self::RECLAIM_TIMEOUT)
+            ->update([
+                'status'      => SystemNoticeTargetBase::STATUS_PENDING,
+                'claim_token' => '',
+                'claim_time'  => 0,
+                'update_time' => time(),
+            ]);
+
+        return 'reclaimed: ' . $affected;
+    }
+}