From ca16cd56de463968d4493fb0a7f19c96b01d9651 Mon Sep 17 00:00:00 2001 From: augushong Date: Mon, 12 Oct 2026 12:41:03 +0000 Subject: [PATCH] =?UTF-8?q?feat(notice):=20timer=20=E5=BC=82=E6=AD=A5?= =?UTF-8?q?=E6=8A=95=E9=80=92=E6=B6=88=E8=B4=B9=E8=80=85=E4=B8=8E=E8=B6=85?= =?UTF-8?q?=E6=97=B6=E5=9B=9E=E6=94=B6=E4=BB=BB=E5=8A=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/tools/controller/timer/NoticeConsumer.php | 11 + app/tools/controller/timer/NoticeReclaim.php | 11 + extend/base/common/command/timer/config.php | 14 ++ .../controller/timer/NoticeConsumerBase.php | 222 ++++++++++++++++++ .../controller/timer/NoticeReclaimBase.php | 56 +++++ 5 files changed, 314 insertions(+) create mode 100644 app/tools/controller/timer/NoticeConsumer.php create mode 100644 app/tools/controller/timer/NoticeReclaim.php create mode 100644 extend/base/tools/controller/timer/NoticeConsumerBase.php create mode 100644 extend/base/tools/controller/timer/NoticeReclaimBase.php diff --git a/app/tools/controller/timer/NoticeConsumer.php b/app/tools/controller/timer/NoticeConsumer.php new file mode 100644 index 0000000..8f4a42f --- /dev/null +++ b/app/tools/controller/timer/NoticeConsumer.php @@ -0,0 +1,11 @@ + 1, 'run_type' => 'all', ], + [ + 'name' => 'notice_consumer', + 'type' => 'site', + 'target' => '/tools/timer.NoticeConsumer/do', + 'frequency' => 10, + 'concurrency' => 2, + ], + [ + 'name' => 'notice_reclaim', + 'type' => 'site', + 'target' => '/tools/timer.NoticeReclaim/do', + 'frequency' => 60, + 'concurrency' => 1, + ], ]; diff --git a/extend/base/tools/controller/timer/NoticeConsumerBase.php b/extend/base/tools/controller/timer/NoticeConsumerBase.php new file mode 100644 index 0000000..4abd231 --- /dev/null +++ b/extend/base/tools/controller/timer/NoticeConsumerBase.php @@ -0,0 +1,222 @@ +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(), + ]); + } +} diff --git a/extend/base/tools/controller/timer/NoticeReclaimBase.php b/extend/base/tools/controller/timer/NoticeReclaimBase.php new file mode 100644 index 0000000..4263f31 --- /dev/null +++ b/extend/base/tools/controller/timer/NoticeReclaimBase.php @@ -0,0 +1,56 @@ +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; + } +}