1.图片任务新增派发锁机制2.新增任务检查和通知

app/Console/Commands/CheckAiTaskHealthCommand.php

@@ -0,0 +1,589 @@
+<?php
+
+namespace App\Console\Commands;
+
+use App\Consts\BaseConst;
+use App\Models\MpGeneratePicTask;
+use Illuminate\Console\Command;
+use Illuminate\Support\Facades\DB;
+use Illuminate\Support\Facades\Redis;
+
+/**
+ * 图片任务健康检查 & 钉钉告警
+ *
+ * 由 app/Console/Kernel.php 每分钟调度一次（ai:check-task-health）。
+ * 每轮采集下面 4 项指标，命中后需要"连续命中 N 次"才真正发出告警，避免瞬时波动误报；
+ * 同一规则告警后进入冷却期，冷却期内不重复发送；指标恢复正常后发送一条"已恢复"通知。
+ *
+ * =========================== 通知规则一览（全部参数见下方常量区） ===========================
+ *
+ * 【规则A】队列积压（QUEUE_BACKLOG_*）
+ *   判定：监控队列的 Redis 长度 >= QUEUE_BACKLOG_THRESHOLD 条，
+ *         且连续 QUEUE_BACKLOG_CONSECUTIVE 次命中（每分钟检查一次）。
+ *   含义：job 已经派发进队列但没人消费——worker 进程挂了/没启动，
+ *         或队列名、Redis 实例与派发方不一致。
+ *
+ * 【规则B】任务长时间未启动（PENDING_*）
+ *   判定：队列派发型模型（GPT-Image2、火山系列）中 status=pending
+ *         且 created_at 早于 PENDING_TIMEOUT_MINUTES 分钟的任务数 > 0，
+ *         且连续 PENDING_CONSECUTIVE 次命中。
+ *   含义：任务已创建却一直没被 worker 取走，用户端会一直显示 0 进度。
+ *   说明：即梦（jimeng_4.0）、NanoBanana 走内部排队/同步提交，等待属于正常，不参与本规则。
+ *
+ * 【规则C】任务长时间处理中（PROCESSING_*）
+ *   判定：status=processing 且 started_at 早于 PROCESSING_TIMEOUT_MINUTES 分钟的任务数 > 0，
+ *         且连续 PROCESSING_CONSECUTIVE 次命中。
+ *   含义：任务已被 worker 取走但卡在第三方接口，或 worker 被强杀导致状态悬挂。
+ *
+ * 【规则D】失败突增（FAILED_*）
+ *   判定：最近 FAILED_WINDOW_MINUTES 分钟内 status=failed 的任务数 > FAILED_THRESHOLD，
+ *         且连续 FAILED_CONSECUTIVE 次命中。
+ *   含义：第三方账号/额度异常（例如历史出现过的 USER_INACTIVE）或内容审核批量拦截。
+ *
+ * 【通用行为】
+ *   1. 同一规则告警后，ALERT_COOLDOWN_MINUTES 分钟内不重复发送；超过冷却期仍异常则重复提醒一次；
+ *   2. 指标恢复正常时发送一条"已恢复"通知（含本次异常持续时长）；
+ *   3. 所有通知都带环境（APP_ENV）、命中详情与排查建议；
+ *   4. 连续次数、冷却状态记录在 Redis（ALERT_KEY_PREFIX + 规则名），ALERT_STATE_TTL 后自动过期；
+ *   5. 钉钉通知直接复用项目已有的 sendNotice()（读取 .env 的 DD_WEB_HOOK）；
+ *      未配置 webhook 或发送失败都只记录日志，不影响命令执行。
+ * ===========================================================================================
+ */
+class CheckAiTaskHealthCommand extends Command
+{
+    /**
+     * 命令名与参数
+     *
+     * @var string
+     */
+    protected $signature = 'ai:check-task-health {--dry-run : 只输出判定结果，不发送钉钉也不写入状态}';
+
+    /**
+     * 命令说明
+     *
+     * @var string
+     */
+    protected $description = '图片任务健康检查（队列积压/任务卡住/失败突增）并在异常时发送钉钉告警';
+
+    // ============================== 规则A：队列积压 ==============================
+    /** 监控的 Redis 队列：队列名 => 展示名（Laravel Redis 队列的 key 为 queues:{队列名}） */
+    const MONITOR_QUEUES = [
+        '{GenerateGptImage2Pics}' => 'GPT-Image2',
+        '{GenerateVolcPics}'      => '火山API',
+    ];
+    /** 队列积压条数阈值（达到即算命中） */
+    const QUEUE_BACKLOG_THRESHOLD = 200;
+    /** 队列积压需要连续命中的次数（每分钟检查一次） */
+    const QUEUE_BACKLOG_CONSECUTIVE = 3;
+
+    // ============================== 规则B：任务长时间未启动 ==============================
+    /** pending 超时时间（分钟），超过即算命中 */
+    const PENDING_TIMEOUT_MINUTES = 10;
+    /** pending 超时需要连续命中的次数 */
+    const PENDING_CONSECUTIVE = 2;
+
+    // ============================== 规则C：任务长时间处理中 ==============================
+    /** processing 超时时间（分钟），超过即算命中 */
+    const PROCESSING_TIMEOUT_MINUTES = 15;
+    /** processing 超时需要连续命中的次数 */
+    const PROCESSING_CONSECUTIVE = 2;
+
+    // ============================== 规则D：失败突增 ==============================
+    /** 失败统计的滑动窗口（分钟） */
+    const FAILED_WINDOW_MINUTES = 10;
+    /** 窗口内失败条数阈值（超过才告警） */
+    const FAILED_THRESHOLD = 20;
+    /** 失败突增需要连续命中的次数 */
+    const FAILED_CONSECUTIVE = 2;
+    /** 失败样例在告警中展示的条数 */
+    const FAILED_SAMPLE_LIMIT = 3;
+
+    // ============================== 通用参数 ==============================
+    /** 同一规则告警后的冷却时间（分钟），冷却期内不重复发送 */
+    const ALERT_COOLDOWN_MINUTES = 30;
+    /** 告警状态在 Redis 中的 key 前缀 */
+    const ALERT_KEY_PREFIX = 'ai_task_alert:';
+    /** 告警状态保留时间（秒），避免 Redis 中残留脏状态 */
+    const ALERT_STATE_TTL = 604800; // 7 天
+    /** 使用的 Redis 连接，必须与队列保持一致 */
+    const ALERT_REDIS_CONNECTION = 'default';
+    /** 展示错误信息时截断的字符数 */
+    const ERROR_MESSAGE_LIMIT = 80;
+
+    /**
+     * 执行命令
+     *
+     * @return int
+     */
+    public function handle()
+    {
+        $dryRun = (bool) $this->option('dry-run');
+
+        $this->info('图片任务健康检查开始' . ($dryRun ? '（dry-run：不发送钉钉、不写入状态）' : ''));
+        $this->line('环境：' . config('app.env') . '　时间：' . now()->toDateTimeString());
+
+        try {
+            // 采集 4 项规则
+            $rules = [
+                $this->buildQueueBacklogRule(),
+                $this->buildPendingTimeoutRule(),
+                $this->buildProcessingTimeoutRule(),
+                $this->buildFailedSpikeRule(),
+            ];
+
+            foreach ($rules as $rule) {
+                $this->processRule($rule, $dryRun);
+            }
+        } catch (\Throwable $e) {
+            // 检查本身异常不影响其它定时任务，只记录并输出
+            $this->error('图片任务健康检查执行失败：' . $e->getMessage());
+            dLog('generate')->error('图片任务健康检查执行失败', ['error' => $e->getMessage()]);
+
+            return 1;
+        }
+
+        $this->info('图片任务健康检查结束');
+
+        return 0;
+    }
+
+    /**
+     * 规则处理：连续计数、冷却、告警与恢复通知、状态落库
+     *
+     * @param array $rule   规则数据
+     * @param bool  $dryRun 是否演练模式（不发通知、不写状态）
+     * @return void
+     */
+    private function processRule(array $rule, bool $dryRun): void
+    {
+        $state = $this->loadState($rule['key']);
+        $cooldownSeconds = self::ALERT_COOLDOWN_MINUTES * 60;
+
+        $this->outputRuleResult($rule, $state);
+
+        if ($rule['triggered']) {
+            $state['consecutive'] = (int) $state['consecutive'] + 1;
+
+            if ($state['consecutive'] === 1 && empty($state['first_hit_at'])) {
+                $state['first_hit_at'] = time();
+            }
+
+            if (!$state['alerting']) {
+                // 未处于告警状态：连续命中达到配置次数才发送
+                if ($state['consecutive'] >= (int) $rule['consecutive']) {
+                    if ($dryRun) {
+                        $this->line('  → [dry-run] 已达到连续命中次数，将发送告警');
+                    } else {
+                        $this->sendAlert($rule, false);
+                        $state['alerting'] = true;
+                        $state['alerted_at'] = time();
+                    }
+                } else {
+                    $this->line('  → 命中 ' . $state['consecutive'] . '/' . $rule['consecutive'] . ' 次，未达连续次数，暂不告警');
+                }
+            } else {
+                // 已在告警中：冷却结束后重复提醒
+                $elapsed = time() - (int) $state['alerted_at'];
+                if ($elapsed >= $cooldownSeconds) {
+                    if ($dryRun) {
+                        $this->line('  → [dry-run] 冷却期已过且异常仍在持续，将重复发送告警');
+                    } else {
+                        $this->sendAlert($rule, true);
+                        $state['alerted_at'] = time();
+                    }
+                } else {
+                    $this->line('  → 告警中，冷却剩余约 ' . (int) ceil(($cooldownSeconds - $elapsed) / 60) . ' 分钟');
+                }
+            }
+        } else {
+            if ($state['alerting']) {
+                if ($dryRun) {
+                    $this->line('  → [dry-run] 指标已恢复，将发送恢复通知');
+                } else {
+                    $this->sendRecovery($rule, $state);
+                }
+                $state['alerting'] = false;
+            }
+            $state['consecutive'] = 0;
+            $state['first_hit_at'] = 0;
+        }
+
+        if (!$dryRun) {
+            $state['last_detail'] = implode(' | ', $rule['detail']);
+            $state['last_check_at'] = time();
+            $this->saveState($rule['key'], $state);
+        }
+    }
+
+    /**
+     * 规则A：队列积压
+     *
+     * @return array
+     */
+    private function buildQueueBacklogRule(): array
+    {
+        $detail = [];
+        $hit = [];
+
+        try {
+            foreach (self::MONITOR_QUEUES as $queue => $label) {
+                $length = (int) Redis::connection(self::ALERT_REDIS_CONNECTION)->llen('queues:' . $queue);
+                $detail[] = $label . '队列（queues:' . $queue . '）：' . $length . ' 条，阈值 ' . self::QUEUE_BACKLOG_THRESHOLD . ' 条';
+
+                if ($length >= self::QUEUE_BACKLOG_THRESHOLD) {
+                    $hit[] = $label . '=' . $length;
+                }
+            }
+        } catch (\Throwable $e) {
+            // Redis 不可用时无法判定队列长度，只提示不告警（避免误报）
+            $detail[] = 'Redis 队列长度读取失败：' . $e->getMessage();
+            $hit = [];
+        }
+
+        return [
+            'key'         => 'queue_backlog',
+            'title'       => '队列积压',
+            'rule_text'   => '队列长度 >= ' . self::QUEUE_BACKLOG_THRESHOLD . ' 条，连续 ' . self::QUEUE_BACKLOG_CONSECUTIVE . ' 次命中',
+            'consecutive' => self::QUEUE_BACKLOG_CONSECUTIVE,
+            'triggered'   => !empty($hit),
+            'detail'      => $detail,
+            'hit_text'    => implode('，', $hit),
+            'suggest'     => '检查 worker 进程是否存活（supervisorctl status generate_gpt_image2_pics:*），确认队列名与 Redis 实例和派发方一致',
+        ];
+    }
+
+    /**
+     * 规则B：任务长时间未启动
+     *
+     * @return array
+     */
+    private function buildPendingTimeoutRule(): array
+    {
+        $deadline = now()->subMinutes(self::PENDING_TIMEOUT_MINUTES);
+        $query = DB::table('mp_generate_pic_tasks')
+            ->where('status', MpGeneratePicTask::STATUS_PENDING)
+            ->whereIn('model', $this->pendingWatchModels())
+            ->where('created_at', '<', $deadline);
+
+        $count = (clone $query)->count();
+        $detail = ['pending 超过 ' . self::PENDING_TIMEOUT_MINUTES . ' 分钟的任务：' . $count . ' 条'];
+
+        if ($count > 0) {
+            $oldest = (clone $query)->orderBy('created_at', 'asc')->first();
+            if ($oldest) {
+                $detail[] = '最早一条：id=' . $oldest->id
+                    . '，模型=' . $oldest->model
+                    . '，外部任务号=' . $oldest->task_id
+                    . '，创建于 ' . $oldest->created_at
+                    . '（已等待 ' . $this->minutesSince($oldest->created_at) . ' 分钟）';
+            }
+        }
+
+        return [
+            'key'         => 'pending_timeout',
+            'title'       => '任务长时间未启动',
+            'rule_text'   => 'pending 超过 ' . self::PENDING_TIMEOUT_MINUTES . ' 分钟的任务数 > 0，连续 ' . self::PENDING_CONSECUTIVE . ' 次命中',
+            'consecutive' => self::PENDING_CONSECUTIVE,
+            'triggered'   => $count > 0,
+            'detail'      => $detail,
+            'hit_text'    => '超时 pending ' . $count . ' 条',
+            'suggest'     => '任务已入队但未被消费，优先检查对应队列的 worker 与队列积压情况',
+        ];
+    }
+
+    /**
+     * 规则C：任务长时间处理中
+     *
+     * @return array
+     */
+    private function buildProcessingTimeoutRule(): array
+    {
+        $deadline = now()->subMinutes(self::PROCESSING_TIMEOUT_MINUTES);
+        $query = DB::table('mp_generate_pic_tasks')
+            ->where('status', MpGeneratePicTask::STATUS_PROCESSING)
+            ->whereNotNull('started_at')
+            ->where('started_at', '<', $deadline);
+
+        $count = (clone $query)->count();
+        $detail = ['processing 超过 ' . self::PROCESSING_TIMEOUT_MINUTES . ' 分钟的任务：' . $count . ' 条'];
+
+        if ($count > 0) {
+            $oldest = (clone $query)->orderBy('started_at', 'asc')->first();
+            if ($oldest) {
+                $detail[] = '最早一条：id=' . $oldest->id
+                    . '，模型=' . $oldest->model
+                    . '，外部任务号=' . $oldest->task_id
+                    . '，开始于 ' . $oldest->started_at
+                    . '（已处理 ' . $this->minutesSince($oldest->started_at) . ' 分钟）';
+            }
+        }
+
+        return [
+            'key'         => 'processing_timeout',
+            'title'       => '任务长时间处理中',
+            'rule_text'   => 'processing 超过 ' . self::PROCESSING_TIMEOUT_MINUTES . ' 分钟的任务数 > 0，连续 ' . self::PROCESSING_CONSECUTIVE . ' 次命中',
+            'consecutive' => self::PROCESSING_CONSECUTIVE,
+            'triggered'   => $count > 0,
+            'detail'      => $detail,
+            'hit_text'    => '超时 processing ' . $count . ' 条',
+            'suggest'     => '检查第三方接口是否超时/限流，以及 worker 是否被强杀（supervisor 的 stopwaitsecs 是否小于任务超时时间）',
+        ];
+    }
+
+    /**
+     * 规则D：失败突增
+     *
+     * @return array
+     */
+    private function buildFailedSpikeRule(): array
+    {
+        $deadline = now()->subMinutes(self::FAILED_WINDOW_MINUTES);
+        $query = DB::table('mp_generate_pic_tasks')
+            ->where('status', MpGeneratePicTask::STATUS_FAILED)
+            ->where('updated_at', '>=', $deadline);
+
+        $count = (clone $query)->count();
+        $detail = ['最近 ' . self::FAILED_WINDOW_MINUTES . ' 分钟内失败的任务：' . $count . ' 条，阈值 > ' . self::FAILED_THRESHOLD . ' 条'];
+
+        if ($count > self::FAILED_THRESHOLD) {
+            $samples = (clone $query)->orderBy('updated_at', 'desc')->limit(self::FAILED_SAMPLE_LIMIT)->get(['id', 'model', 'error_message', 'updated_at']);
+            foreach ($samples as $sample) {
+                $detail[] = '样例：id=' . $sample->id
+                    . '，模型=' . $sample->model
+                    . '，时间=' . $sample->updated_at
+                    . '，错误=' . mb_substr((string) $sample->error_message, 0, self::ERROR_MESSAGE_LIMIT);
+            }
+        }
+
+        return [
+            'key'         => 'failed_spike',
+            'title'       => '失败任务突增',
+            'rule_text'   => '最近 ' . self::FAILED_WINDOW_MINUTES . ' 分钟内失败数 > ' . self::FAILED_THRESHOLD . ' 条，连续 ' . self::FAILED_CONSECUTIVE . ' 次命中',
+            'consecutive' => self::FAILED_CONSECUTIVE,
+            'triggered'   => $count > self::FAILED_THRESHOLD,
+            'detail'      => $detail,
+            'hit_text'    => '失败 ' . $count . ' 条',
+            'suggest'     => '检查第三方账号/额度状态与内容审核拦截，确认接口返回的错误码',
+        ];
+    }
+
+    /**
+     * pending 规则监控的模型：仅"队列派发型"（GPT-Image2、火山系列）
+     *
+     * 即梦（jimeng_4.0）、NanoBanana 走内部排队/同步提交，长时间 pending 属正常，不参与告警。
+     *
+     * @return array
+     */
+    private function pendingWatchModels(): array
+    {
+        return array_merge(BaseConst::GPT_IMAGE2_MODELS, BaseConst::VOLC_PIC_MODELS);
+    }
+
+    /**
+     * 控制台输出单条规则的判定结果
+     *
+     * @param array $rule
+     * @param array $state
+     * @return void
+     */
+    private function outputRuleResult(array $rule, array $state): void
+    {
+        $status = $rule['triggered']
+            ? '命中（连续 ' . ((int) $state['consecutive'] + 1) . '/' . $rule['consecutive'] . '）'
+            : '正常';
+
+        $this->line('【' . $rule['title'] . '】' . $status);
+
+        foreach ($rule['detail'] as $line) {
+            $this->line('  - ' . $line);
+        }
+    }
+
+    /**
+     * 发送告警（钉钉 + 控制台 + 日志）
+     *
+     * @param array $rule
+     * @param bool  $repeat 是否为持续告警的重复提醒
+     * @return void
+     */
+    private function sendAlert(array $rule, bool $repeat): void
+    {
+        $lines = [
+            ($repeat ? '【图片任务告警·持续】' : '【图片任务告警】') . $rule['title'],
+            '环境：' . config('app.env'),
+            '规则：' . $rule['rule_text'],
+            '当前：' . $rule['hit_text'],
+        ];
+
+        foreach ($rule['detail'] as $line) {
+            $lines[] = '  - ' . $line;
+        }
+
+        if ($repeat) {
+            $lines[] = '说明：该异常仍在持续，距上次通知已 ' . self::ALERT_COOLDOWN_MINUTES . ' 分钟';
+        }
+
+        $lines[] = '建议：' . $rule['suggest'];
+        $lines[] = '时间：' . now()->toDateTimeString();
+
+        $content = implode("\n", $lines);
+
+        $this->error($content);
+        dLog('generate')->error('图片任务健康告警：' . $rule['title'], [
+            'rule'   => $rule['key'],
+            'detail' => $rule['detail'],
+        ]);
+        logDB('generate', 'error', '图片任务健康告警：' . $rule['title'], [
+            'rule'   => $rule['key'],
+            'detail' => $rule['detail'],
+        ]);
+
+        $this->notify($content);
+    }
+
+    /**
+     * 发送恢复通知（只在上一轮处于告警状态时发送）
+     *
+     * @param array $rule
+     * @param array $state
+     * @return void
+     */
+    private function sendRecovery(array $rule, array $state): void
+    {
+        $lines = [
+            '【图片任务恢复】' . $rule['title'] . ' 已恢复正常',
+            '环境：' . config('app.env'),
+        ];
+
+        if (!empty($state['first_hit_at'])) {
+            $lines[] = '本次异常持续约 ' . (int) ceil((time() - (int) $state['first_hit_at']) / 60) . ' 分钟';
+        }
+
+        foreach ($rule['detail'] as $line) {
+            $lines[] = '  - ' . $line;
+        }
+
+        $lines[] = '时间：' . now()->toDateTimeString();
+
+        $content = implode("\n", $lines);
+
+        $this->info($content);
+        dLog('generate')->info('图片任务健康恢复：' . $rule['title'], ['rule' => $rule['key']]);
+        logDB('generate', 'info', '图片任务健康恢复：' . $rule['title'], ['rule' => $rule['key']]);
+
+        $this->notify($content);
+    }
+
+    /**
+     * 发送钉钉通知
+     *
+     * 直接复用项目已有的 sendNotice()（读取线上 .env 的 DD_WEB_HOOK，默认 @所有人）。
+     * webhook 未配置或发送失败都只记录日志，绝不影响检查命令本身。
+     *
+     * @param string $content
+     * @return void
+     */
+    private function notify(string $content): void
+    {
+        if (empty(env('DD_WEB_HOOK'))) {
+            $this->warn('DD_WEB_HOOK 未配置，跳过钉钉发送');
+            dLog('generate')->warning('图片任务健康告警未发送：DD_WEB_HOOK 未配置', ['content' => $content]);
+            return;
+        }
+
+        try {
+            sendNotice($content);
+        } catch (\Throwable $e) {
+            $this->warn('钉钉发送失败：' . $e->getMessage());
+            dLog('generate')->error('图片任务健康告警发送失败', [
+                'error'   => $e->getMessage(),
+                'content' => $content,
+            ]);
+        }
+    }
+
+    /**
+     * 读取规则状态（连续命中次数、是否告警中、首次命中时间等）
+     *
+     * @param string $ruleKey
+     * @return array
+     */
+    private function loadState(string $ruleKey): array
+    {
+        $default = [
+            'consecutive' => 0,
+            'alerting'    => false,
+            'alerted_at'  => 0,
+            'first_hit_at'=> 0,
+        ];
+
+        try {
+            $raw = Redis::connection(self::ALERT_REDIS_CONNECTION)->get($this->stateKey($ruleKey));
+            if (empty($raw)) {
+                return $default;
+            }
+
+            $state = json_decode($raw, true);
+            if (!is_array($state)) {
+                return $default;
+            }
+
+            return array_merge($default, $state);
+        } catch (\Throwable $e) {
+            dLog('generate')->warning('读取图片任务告警状态失败', [
+                'rule'  => $ruleKey,
+                'error' => $e->getMessage(),
+            ]);
+
+            return $default;
+        }
+    }
+
+    /**
+     * 保存规则状态
+     *
+     * @param string $ruleKey
+     * @param array  $state
+     * @return void
+     */
+    private function saveState(string $ruleKey, array $state): void
+    {
+        try {
+            Redis::connection(self::ALERT_REDIS_CONNECTION)->set(
+                $this->stateKey($ruleKey),
+                json_encode($state, JSON_UNESCAPED_UNICODE),
+                'EX',
+                self::ALERT_STATE_TTL
+            );
+        } catch (\Throwable $e) {
+            dLog('generate')->warning('保存图片任务告警状态失败', [
+                'rule'  => $ruleKey,
+                'error' => $e->getMessage(),
+            ]);
+        }
+    }
+
+    /**
+     * 规则状态在 Redis 中的 key
+     *
+     * @param string $ruleKey
+     * @return string
+     */
+    private function stateKey(string $ruleKey): string
+    {
+        return self::ALERT_KEY_PREFIX . 'state:' . $ruleKey;
+    }
+
+    /**
+     * 计算某个时间点距今经过的分钟数
+     *
+     * @param string $dateTime
+     * @return int
+     */
+    private function minutesSince(string $dateTime): int
+    {
+        $timestamp = strtotime($dateTime);
+
+        return $timestamp ? (int) floor((time() - $timestamp) / 60) : 0;
+    }
+}

app/Console/Kernel.php

@@ -59,6 +59,9 @@ class Kernel extends ConsoleKernel
         // 同步任务中心状态和结果（关联图片/视频任务）
         $schedule->command('taskCenter:sync')->everyMinute()->withoutOverlapping(5);
 
+        // 图片任务健康检查与告警（队列积压/任务长时间未完成/失败突增，异常时钉钉通知）
+        $schedule->command('ai:check-task-health')->everyMinute()->withoutOverlapping(5);
+
         // 积分消耗与token用量日统计（聚合前一天明细）
         $schedule->command('stats:points-daily')->dailyAt('00:10')->withoutOverlapping();
     }

app/Jobs/ProcessGptImage2GenerationJob.php

@@ -55,6 +55,9 @@ class ProcessGptImage2GenerationJob implements ShouldQueue
      */
     public function handle(AIImageGenerationService $service)
     {
+        // 本次 job 是否已正常结束；异常时不释放派发锁，交给队列重试或 failed() 释放
+        $finished = false;
+
         try {
             // 查找任务
             $task = MpGeneratePicTask::find($this->taskId);
@@ -62,6 +65,7 @@ class ProcessGptImage2GenerationJob implements ShouldQueue
             if (!$task) {
                 dLog('generate')->error('队列任务：找不到GPT-Image2图片生成任务', ['task_id' => $this->taskId]);
                 logDB('generate', 'error', '队列任务：找不到GPT-Image2图片生成任务', ['task_id' => $this->taskId]);
+                $finished = true;
                 return;
             }
 
@@ -71,6 +75,7 @@ class ProcessGptImage2GenerationJob implements ShouldQueue
                     'task_id' => $this->taskId,
                     'status' => $task->status
                 ]);
+                $finished = true;
                 return;
             }
 
@@ -80,6 +85,7 @@ class ProcessGptImage2GenerationJob implements ShouldQueue
             $service->submitTaskToGptImage2Api($task);
 
             dLog('generate')->info('队列任务：GPT-Image2图片生成任务处理完成', ['task_id' => $this->taskId]);
+            $finished = true;
 
         } catch (\Exception $e) {
             dLog('generate')->error('队列任务：处理GPT-Image2图片生成任务失败', [
@@ -92,8 +98,16 @@ class ProcessGptImage2GenerationJob implements ShouldQueue
                 'trace' => $e->getTraceAsString()
             ]);
 
-            // 重新抛出异常，让队列系统处理重试
+            // 重新抛出异常，让队列系统处理重试；此处不释放派发锁，
+            // 避免重试期间任务被重复派发，最终失败时由 failed() 释放
             throw $e;
+        } finally {
+            if ($finished) {
+                AIImageGenerationService::releaseTaskDispatchLock(
+                    AIImageGenerationService::DISPATCH_GROUP_GPT_IMAGE2,
+                    $this->taskId
+                );
+            }
         }
     }
 
@@ -129,5 +143,11 @@ class ProcessGptImage2GenerationJob implements ShouldQueue
                 'error' => $e->getMessage()
             ]);
         }
+
+        // 最终失败：释放派发锁，使任务在下一次扫描时可以重新派发
+        AIImageGenerationService::releaseTaskDispatchLock(
+            AIImageGenerationService::DISPATCH_GROUP_GPT_IMAGE2,
+            $this->taskId
+        );
     }
 }

app/Jobs/ProcessVolcImageGenerationJob.php

@@ -55,6 +55,9 @@ class ProcessVolcImageGenerationJob implements ShouldQueue
      */
     public function handle(AIImageGenerationService $service)
     {
+        // 本次 job 是否已正常结束；异常时不释放派发锁，交给队列重试或 failed() 释放
+        $finished = false;
+
         try {
             // 查找任务
             $task = MpGeneratePicTask::find($this->taskId);
@@ -62,6 +65,7 @@ class ProcessVolcImageGenerationJob implements ShouldQueue
             if (!$task) {
                 dLog('generate')->error('队列任务：找不到图片生成任务', ['task_id' => $this->taskId]);
                 logDB('generate', 'error', '队列任务：找不到图片生成任务', ['task_id' => $this->taskId]);
+                $finished = true;
                 return;
             }
 
@@ -71,6 +75,7 @@ class ProcessVolcImageGenerationJob implements ShouldQueue
                     'task_id' => $this->taskId,
                     'status' => $task->status
                 ]);
+                $finished = true;
                 return;
             }
 
@@ -80,6 +85,7 @@ class ProcessVolcImageGenerationJob implements ShouldQueue
             $service->submitTaskToVolcApi($task);
 
             dLog('generate')->info('队列任务：火山API图片生成任务处理完成', ['task_id' => $this->taskId]);
+            $finished = true;
 
         } catch (\Exception $e) {
             dLog('generate')->error('队列任务：处理火山API图片生成任务失败', [
@@ -92,8 +98,16 @@ class ProcessVolcImageGenerationJob implements ShouldQueue
                 'trace' => $e->getTraceAsString()
             ]);
 
-            // 重新抛出异常，让队列系统处理重试
+            // 重新抛出异常，让队列系统处理重试；此处不释放派发锁，
+            // 避免重试期间任务被重复派发，最终失败时由 failed() 释放
             throw $e;
+        } finally {
+            if ($finished) {
+                AIImageGenerationService::releaseTaskDispatchLock(
+                    AIImageGenerationService::DISPATCH_GROUP_VOLC,
+                    $this->taskId
+                );
+            }
         }
     }
 
@@ -129,5 +143,11 @@ class ProcessVolcImageGenerationJob implements ShouldQueue
                 'error' => $e->getMessage()
             ]);
         }
+
+        // 最终失败：释放派发锁，使任务在下一次扫描时可以重新派发
+        AIImageGenerationService::releaseTaskDispatchLock(
+            AIImageGenerationService::DISPATCH_GROUP_VOLC,
+            $this->taskId
+        );
     }
 }

app/Services/AIGeneration/AIImageGenerationService.php

@@ -27,6 +27,22 @@ class AIImageGenerationService
     const MODEL_NANO_BANANA_PRO = 'NanoBananaPro';
     const MODEL_GPT_IMAGE_2 = 'GptImage2';
 
+    // 队列派发去重相关常量
+    /** 派发锁在 Redis 中的 key 前缀 */
+    const DISPATCH_LOCK_PREFIX = 'ai_pic_dispatch:';
+    /** 派发锁使用的 Redis 连接，与队列保持一致 */
+    const DISPATCH_LOCK_CONNECTION = 'default';
+    /** 派发锁有效期（秒）：worker 长时间未消费时的兜底重派时间 */
+    const DISPATCH_LOCK_TTL = 600;
+    /** 每轮最多派发的任务数，避免一次性推送过多任务 */
+    const DISPATCH_BATCH_LIMIT = 50;
+    /** 每轮扫描的候选任务数，需大于 DISPATCH_BATCH_LIMIT，避免已入队的任务挡住后面的任务 */
+    const DISPATCH_CANDIDATE_LIMIT = 500;
+    /** 派发锁分组：火山API */
+    const DISPATCH_GROUP_VOLC = 'volc_pics';
+    /** 派发锁分组：GPT-Image2 */
+    const DISPATCH_GROUP_GPT_IMAGE2 = 'gpt_image2_pics';
+
     public function __construct(VolcEngineService $volcEngineService, PointsService $pointsService)
     {
         $this->volcEngineService = $volcEngineService;
@@ -1471,28 +1487,44 @@ class AIImageGenerationService
         $volcPendingTasks = MpGeneratePicTask::where('status', MpGeneratePicTask::STATUS_PENDING)
             ->whereIn('model', BaseConst::VOLC_PIC_MODELS)
             ->orderBy('created_at', 'asc')
-            ->limit(50) // 每次最多处理50个任务，避免超时
+            ->limit(self::DISPATCH_CANDIDATE_LIMIT) // 候选窗口需大于每轮派发上限，避免已入队的任务挡住后面的任务
             ->get();
 
+        $volcDispatched = 0;
         foreach ($volcPendingTasks as $task) {
+            if ($volcDispatched >= self::DISPATCH_BATCH_LIMIT) {
+                break;
+            }
+
+            // 同一任务已在队列中或正在执行时跳过，避免重复入队
+            if (!$this->dispatchTaskOnce(self::DISPATCH_GROUP_VOLC, (int)$task->id, \App\Jobs\ProcessVolcImageGenerationJob::class)) {
+                continue;
+            }
+
             dLog('generate')->info('开始分发火山API任务到队列: ' . $task->task_id);
-            
-            // 分发到队列
-            \App\Jobs\ProcessVolcImageGenerationJob::dispatch($task->id)->onConnection('redis');
+            $volcDispatched++;
         }
         
         // 2. 处理GPT-Image2的pending任务（批量提交到队列）
         $gptImage2PendingTasks = MpGeneratePicTask::where('status', MpGeneratePicTask::STATUS_PENDING)
             ->whereIn('model', BaseConst::GPT_IMAGE2_MODELS)
             ->orderBy('created_at', 'asc')
-            ->limit(50) // 每次最多处理50个任务，避免超时
+            ->limit(self::DISPATCH_CANDIDATE_LIMIT) // 候选窗口需大于每轮派发上限，避免已入队的任务挡住后面的任务
             ->get();
 
+        $gptImage2Dispatched = 0;
         foreach ($gptImage2PendingTasks as $task) {
+            if ($gptImage2Dispatched >= self::DISPATCH_BATCH_LIMIT) {
+                break;
+            }
+
+            // 同一任务已在队列中或正在执行时跳过，避免重复入队
+            if (!$this->dispatchTaskOnce(self::DISPATCH_GROUP_GPT_IMAGE2, (int)$task->id, \App\Jobs\ProcessGptImage2GenerationJob::class)) {
+                continue;
+            }
+
             dLog('generate')->info('开始分发GPT-Image2任务到队列: ' . $task->task_id);
-            
-            // 分发到队列
-            \App\Jobs\ProcessGptImage2GenerationJob::dispatch($task->id)->onConnection('redis');
+            $gptImage2Dispatched++;
         }
         
         // 3. 获取所有处理中的任务（即梦AI和NanoBanana系列需要查询状态）
@@ -1538,6 +1570,107 @@ class AIImageGenerationService
     }
 
     /**
+     * 派发单个任务到队列（带派发去重）
+     *
+     * 同一任务在「已入队等待执行」或「正在执行」期间不会重复入队，
+     * 避免每轮扫描（约3秒一次）把同一条任务重复推送进队列。
+     *
+     * @param string $group    派发分组，用于区分同一任务ID在不同业务下的锁
+     * @param int    $taskId   图片任务ID
+     * @param string $jobClass 队列任务类
+     * @return bool 本次是否成功入队；false 表示已有同任务的 job 在队列中或执行中
+     */
+    private function dispatchTaskOnce(string $group, int $taskId, string $jobClass): bool
+    {
+        if (!self::acquireTaskDispatchLock($group, $taskId)) {
+            return false;
+        }
+
+        try {
+            $jobClass::dispatch($taskId)->onConnection('redis');
+        } catch (\Throwable $e) {
+            // 入队失败必须释放锁，否则该任务在锁过期前无法再次派发
+            self::releaseTaskDispatchLock($group, $taskId);
+            throw $e;
+        }
+
+        return true;
+    }
+
+    /**
+     * 获取任务派发锁
+     *
+     * 在队列所在的 Redis 连接上执行原生 `SET key value EX ttl NX` 原子命令，
+     * 不依赖 Lua 脚本（部分 Redis 会禁用 EVAL）也不需要额外的缓存 store 配置。
+     *
+     * @param string $group
+     * @param int    $taskId
+     * @return bool true 表示获取成功（可以入队）
+     */
+    public static function acquireTaskDispatchLock(string $group, int $taskId): bool
+    {
+        try {
+            $result = Redis::connection(self::DISPATCH_LOCK_CONNECTION)->set(
+                self::buildTaskDispatchLockKey($group, $taskId),
+                1,
+                'EX',
+                self::DISPATCH_LOCK_TTL,
+                'NX'
+            );
+
+            // NX 未命中时：predis 返回 null，phpredis 返回 false
+            return $result !== null && $result !== false;
+        } catch (\Throwable $e) {
+            // 锁不可用时降级为不去重（与旧逻辑一致），避免任务因此完全无法派发
+            dLog('generate')->warning('获取图片任务派发锁失败，本次不拦截重复派发', [
+                'group'   => $group,
+                'task_id' => $taskId,
+                'error'   => $e->getMessage(),
+            ]);
+
+            return true;
+        }
+    }
+
+    /**
+     * 释放任务派发锁
+     *
+     * 在任务执行结束、任务已由其它流程处理、任务最终失败时调用；
+     * 未释放的锁会在 DISPATCH_LOCK_TTL 后自动过期，保证任务仍能被重新派发。
+     *
+     * @param string $group
+     * @param int    $taskId
+     * @return void
+     */
+    public static function releaseTaskDispatchLock(string $group, int $taskId): void
+    {
+        try {
+            Redis::connection(self::DISPATCH_LOCK_CONNECTION)->del(
+                self::buildTaskDispatchLockKey($group, $taskId)
+            );
+        } catch (\Throwable $e) {
+            // 释放失败不影响主流程，锁会由 TTL 自动过期
+            dLog('generate')->warning('释放图片任务派发锁失败', [
+                'group'   => $group,
+                'task_id' => $taskId,
+                'error'   => $e->getMessage(),
+            ]);
+        }
+    }
+
+    /**
+     * 构建任务派发锁 key
+     *
+     * @param string $group
+     * @param int    $taskId
+     * @return string
+     */
+    public static function buildTaskDispatchLockKey(string $group, int $taskId): string
+    {
+        return self::DISPATCH_LOCK_PREFIX . $group . ':' . $taskId;
+    }
+
+    /**
      * 处理下一个排队的任务
      *
      * @return void