refactor(api-v13): mq worker 不再自建队列，处理完一律 ack，支持循环上限退出

升级 mq 第一阶段：worker 能跑、能循环到顶自动退出；第二阶段再重构 worker。

- Mq::worker：只做 passive 检查，队列不存在报错退出（先跑 app:create-queue）；
  去掉 exchange 声明与绑定及 $exchange 参数（publish 走默认交换机，router 从未被使用）；
  改为手动 ack + prefetch 1，成功失败一律 ack，循环到顶退出时未处理的消息不再丢失；
  处理满 mint.mq.consumer_loop_limit 条（0 = 不限）或检测到 .stop 后退出
- RabbitMQService：去掉死信、重试、延迟队列，只保留建队列 / 发布 / 阻塞消费
- config：mq.loop_limit → mq.consumer_loop_limit（MQ_CONSUMER_LOOP_LIMIT），
  rabbitmq.queues 精简为业务在用的 8 个，删掉死信与 heartbeat_queue 配置
- SentenceController@update：content 队列无消费者，停发并留 FIXME，同时从队列列表移除
- documents/tasks.md：补 mq:pr，写明建队列、ack 与退出约定

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

api-v13/.env.example

@@ -47,6 +47,9 @@ REDIS_HOST=127.0.0.1
 REDIS_PASSWORD=null
 REDIS_PORT=6379
 
+# mq worker 处理满这么多条消息后退出（由外部重新拉起），0 = 不限
+MQ_CONSUMER_LOOP_LIMIT=0
+
 MAIL_MAILER=log
 MAIL_SCHEME=null
 MAIL_HOST=127.0.0.1

api-v13/app/Console/Commands/MqAiTranslate.php

@@ -47,11 +47,10 @@ class MqAiTranslate extends Command
         if (Tools::isStop()) {
             return 0;
         }
-        $exchange = 'router';
         $queue = 'ai_translate';
         $this->info(" [*] Waiting for {$queue}. To exit press CTRL+C");
         Log::debug("mq worker {$queue} start.");
-        Mq::worker($exchange, $queue, function ($messages, $messageId) use ($queue) {
+        Mq::worker($queue, function ($messages, $messageId) use ($queue) {
             Log::debug('ai translate start', ['message' => count($messages)]);
             $this->info('ai translate task start task='.count($messages));
             if (! is_array($messages) || count($messages) === 0) {

api-v13/app/Console/Commands/MqDiscussion.php

@@ -56,11 +56,10 @@ class MqDiscussion extends Command
         if (Tools::isStop()) {
             return 0;
         }
-        $exchange = 'router';
         $queue = 'discussion';
         $this->info(" [*] Waiting for {$queue}. To exit press CTRL+C");
         Log::info('discussion worker start .');
-        Mq::worker($exchange, $queue, function ($message) {
+        Mq::worker($queue, function ($message) {
             Log::info('mq discussion receive {message}', ['message' => json_encode($message, JSON_UNESCAPED_UNICODE)]);
             $result = 0;
             $msgParam = [];

api-v13/app/Console/Commands/MqEmpty.php

@@ -43,12 +43,10 @@ class MqEmpty extends Command
         if (Tools::isStop()) {
             return 0;
         }
-        $exchange = 'router';
         $queue = 'ai_translate';
         $this->info(" [*] Waiting for {$queue}. To exit press CTRL+C");
         Log::debug("mq worker {$queue} start.");
         Mq::worker(
-            $exchange,
             $queue,
             function ($message) {
                 $this->info('new message');

api-v13/app/Console/Commands/MqExport.php

@@ -40,11 +40,10 @@ class MqExport extends Command
      */
     public function handle()
     {
-        $exchange = 'router';
         $queue = 'export';
         $this->info(" [*] Waiting for {$queue}. To exit press CTRL+C");
         Log::debug('mq:export start.');
-        Mq::worker($exchange, $queue, function ($message) {
+        Mq::worker($queue, function ($message) {
             $data = [
                 'book' => $message->book,
                 'para' => $message->para,

api-v13/app/Console/Commands/MqExportArticle.php

@@ -40,11 +40,10 @@ class MqExportArticle extends Command
      */
     public function handle()
     {
-        $exchange = 'router';
         $queue = 'export_article';
         $this->info(" [*] Waiting for {$queue}. To exit press CTRL+C");
         Log::debug('mq:export_article start.');
-        Mq::worker($exchange, $queue, function ($message) {
+        Mq::worker($queue, function ($message) {
             $data = [
                 'id' => $message->id,
                 '--format' => $message->format,

api-v13/app/Console/Commands/MqExportPaliChapter.php

@@ -40,11 +40,10 @@ class MqExportPaliChapter extends Command
      */
     public function handle()
     {
-        $exchange = 'router';
         $queue = 'export_pali_chapter';
         $this->info(" [*] Waiting for {$queue}. To exit press CTRL+C");
         Log::debug('mq:export_pali_chapter start.');
-        Mq::worker($exchange, $queue, function ($message) {
+        Mq::worker($queue, function ($message) {
             $data = [
                 'book' => $message->book,
                 'para' => $message->para,

api-v13/app/Console/Commands/MqIssues.php

@@ -43,10 +43,9 @@ class MqIssues extends Command
         if (Tools::isStop()) {
             return 0;
         }
-        $exchange = 'router';
         $queue = 'issues';
         $this->info(" [*] Waiting for {$queue}. To exit press CTRL+C");
-        Mq::worker($exchange, $queue, function ($message) {
+        Mq::worker($queue, function ($message) {
             print_r($message);
 
             return 0;

api-v13/app/Console/Commands/MqPr.php

@@ -51,11 +51,10 @@ class MqPr extends Command
         if (Tools::isStop()) {
             return 0;
         }
-        $exchange = 'router';
         $queue = 'suggestion';
         $this->info(" [*] Waiting for {$queue}. Ver. ".$this->ver);
         Log::debug('mq:pr start. ver='.$this->ver);
-        Mq::worker($exchange, $queue, function ($message) {
+        Mq::worker($queue, function ($message) {
             /**生成消息内容 */
 
             $msgTitle = '修改建议';

api-v13/app/Console/Commands/MqProgress.php

@@ -42,13 +42,14 @@ class MqProgress extends Command
     public function handle()
     {
         if (Tools::isStop()) {
+            Log::info('mq:progress. .stop exit');
+
             return 0;
         }
-        $exchange = 'router';
         $queue = 'progress';
         $this->info(" [*] Waiting for {$queue}. To exit press CTRL+C");
         Log::debug('mq:progress start.');
-        Mq::worker($exchange, $queue, function ($message) {
+        Mq::worker($queue, function ($message) {
             $data = [
                 '--book' => $message->book,
                 '--para' => $message->para,
@@ -62,13 +63,11 @@ class MqProgress extends Command
             if ($ok2 !== 0) {
                 Log::error('mq:progress upgrade:progress.chapter fail', $data);
             }
-            $this->info('Received book='.$message->book.' progress='.$ok1.' chapter='.$ok2);
             Log::debug('mq:progress: done book='.$message->book.' progress='.$ok1.' chapter='.$ok2);
 
             return $ok1 + $ok2;
         });
 
         return 0;
-
     }
 }

api-v13/app/Console/Commands/MqTask.php

@@ -44,10 +44,9 @@ class MqTask extends Command
         }
         $this->info('env='.env('RABBITMQ_HOST'));
         $this->info('config='.config('queue.connections.rabbitmq.host'));
-        $exchange = 'router';
         $queue = 'task';
         $this->info(" [*] Waiting for {$queue}. To exit press CTRL+C");
-        Mq::worker($exchange, $queue, function ($message) {
+        Mq::worker($queue, function ($message) {
             $message = json_decode(json_encode($message), true);
             $this->info('name=', $message['name']);
 

api-v13/app/Console/Commands/MqWbwAnalyses.php

@@ -43,11 +43,10 @@ class MqWbwAnalyses extends Command
         if (Tools::isStop()) {
             return 0;
         }
-        $exchange = 'router';
         $queue = 'wbw-analyses';
         $this->info(" [*] Waiting for {$queue}. To exit press CTRL+C");
         Log::debug('mq:wbw.analyses start.');
-        Mq::worker($exchange, $queue, function ($message) {
+        Mq::worker($queue, function ($message) {
             $data = ['id' => implode(',', $message)];
             $ok = $this->call('upgrade:wbw.analyses', $data);
             if ($ok === 0) {

api-v13/app/Http/Api/Mq.php

@@ -5,11 +5,8 @@ namespace App\Http\Api;
 use App\Tools\Tools;
 use Illuminate\Support\Facades\Log;
 use Illuminate\Support\Str;
-use PhpAmqpLib\Channel\AMQPChannel;
-use PhpAmqpLib\Connection\AbstractConnection;
 use PhpAmqpLib\Connection\AMQPStreamConnection;
-use PhpAmqpLib\Exception\AMQPTimeoutException;
-use PhpAmqpLib\Exchange\AMQPExchangeType;
+use PhpAmqpLib\Exception\AMQPProtocolChannelException;
 use PhpAmqpLib\Message\AMQPMessage;
 
 class Mq
@@ -71,14 +68,19 @@ class Mq
     }
 
     /**
-     * @param  string  $exchange
-     * @param  string  $queue
-     * @param  callable|null  $callback
+     * 阻塞消费一个已存在的队列。
+     *
+     * - 队列必须事先建好（`php artisan app:create-queue`），不存在就记日志并抛异常退出，worker 不再自建队列。
+     * - 手动 ack、prefetch 1：每条消息处理结束无论成功失败一律 ack，不重试、不进死信队列。
+     * - 处理满 `mint.mq.consumer_loop_limit` 条（0 = 不限）或检测到 `.stop` 文件后退出。
+     *
+     * @param  callable(mixed $data, ?string $messageId): int|null  $callback  返回非 0 视为失败（只记日志）
+     *
+     * @throws \RuntimeException 队列不存在
      */
-    public static function worker($exchange, $queue, $callback = null)
+    public static function worker(string $queue, ?callable $callback = null): void
     {
-
-        $consumerTag = 'consumer';
+        $consumerTag = '';
 
         $host = config('queue.connections.rabbitmq.host');
         $port = config('queue.connections.rabbitmq.port');
@@ -89,120 +91,68 @@ class Mq
 
         $channel = $connection->channel();
 
-        /*
-     The following code is the same both in the consumer and the producer.
-     In this way we are sure we always have a queue to consume from and an
-         exchange where to publish messages.
- */
-
-        /*
-     name: $queue
-     passive: false
-     durable: true // the queue will survive server restarts
-     exclusive: false // the queue can be accessed in other channels
-     auto_delete: false //the queue won't be deleted once the channel is closed.
- */
-        $channel->queue_declare($queue, false, true, false, false);
-
-        /*
-            name: $exchange
-            type: direct
-            passive: false
-            durable: true // the exchange will survive server restarts
-            auto_delete: false //the exchange won't be deleted once the channel is closed.
-        */
-
-        $channel->exchange_declare($exchange, AMQPExchangeType::DIRECT, false, true, false);
-        $channel->queue_bind($queue, $exchange);
-
-        /**
-         * @param  AMQPMessage  $message
-         */
-        $process_message = function (AMQPMessage $message) use ($callback, $queue) {
+        // passive：只检查不创建。不存在时服务端返回 404 并关闭 channel
+        try {
+            $channel->queue_declare($queue, true);
+        } catch (AMQPProtocolChannelException $e) {
+            Log::error("mq worker: queue {$queue} not found", ['error' => $e->getMessage()]);
+            $connection->close();
+
+            throw new \RuntimeException("mq queue [{$queue}] does not exist, run `php artisan app:create-queue` first", 0, $e);
+        }
+
+        $loopLimit = (int) config('mint.mq.consumer_loop_limit', 0);
+        $processed = 0;
+
+        $process_message = function (AMQPMessage $message) use ($callback, $queue, $loopLimit, &$processed) {
+            $messageId = $message->has('message_id') ? $message->get('message_id') : null;
             Log::debug('received message', [
-                'message_id' => $message->get('message_id'),
-                'content_type' => $message->get('content_type'),
+                'message_id' => $messageId,
+                'content_type' => $message->has('content_type') ? $message->get('content_type') : null,
             ]);
-            if ($callback !== null) {
-                try {
-                    $result = $callback(json_decode($message->getBody()), $message->get('message_id'));
-                    Log::debug(
-                        'mq done',
-                        [
-                            'message_id' => $message->get('message_id'),
-                        ]
-                    );
+
+            try {
+                if ($callback !== null) {
+                    $result = $callback(json_decode($message->getBody()), $messageId);
+                    Log::debug('mq done', ['message_id' => $messageId]);
                     if ($result !== 0) {
                         throw new \Exception('task error');
                     }
-                } catch (\Exception $e) {
-                    Log::error("mq worker {$queue} exception", [
-                        'queue' => $queue,
-                        'message_id' => $message->get('message_id'),
-                        'exception' => $e,
-                    ]);
-                }
-
-                if (Tools::isStop()) {
-                    Log::info('mq worker: .stop file exist. cancel the consumer.');
-                    $message->getChannel()->basic_cancel($message->getConsumerTag());
                 }
+            } catch (\Throwable $e) {
+                Log::error("mq worker {$queue} exception", [
+                    'queue' => $queue,
+                    'message_id' => $messageId,
+                    'exception' => $e,
+                ]);
+            } finally {
+                $message->ack();
             }
 
-            // exit
-            foreach (config('mint.mq.loop_limit') as $key => $value) {
-                if ($queue === $key) {
-                    if ($value > 0) {
-                        if (isset($GLOBALS[$key])) {
-                            $GLOBALS[$key]++;
-                        } else {
-                            $GLOBALS[$key] = 1;
-                        }
-                        if ($GLOBALS[$key] >= $value) {
-                            Log::info("mq exit queue={$queue} loop=".$GLOBALS[$key]);
-                            $message->getChannel()->basic_cancel($message->getConsumerTag());
-                        }
-                    }
-                }
-            }
-            // Send a message with the string "quit" to cancel the consumer.
-            /*
-            if ($message->body === 'quit') {
+            $processed++;
+
+            if ($loopLimit > 0 && $processed >= $loopLimit) {
+                Log::info("mq exit queue={$queue} loop={$processed}");
+                $message->getChannel()->basic_cancel($message->getConsumerTag());
+            } elseif (Tools::isStop()) {
+                Log::info('mq worker: .stop file exist. cancel the consumer.');
                 $message->getChannel()->basic_cancel($message->getConsumerTag());
             }
-            */
-        };
-
-        /*
-            queue: Queue from where to get the messages
-            consumer_tag: Consumer identifier
-            no_local: Don't receive messages published by this consumer.
-            no_ack: If set to true, automatic acknowledgement mode will be used by this consumer. See https://www.rabbitmq.com/confirms.html for details.
-            exclusive: Request exclusive consumer access, meaning only this consumer can access the queue
-            nowait:
-            callback: A PHP Callback
-        */
-        $channel->basic_consume($queue, $consumerTag, false, true, false, false, $process_message);
-
-        /**
-         * @param  AMQPChannel  $channel
-         * @param  AbstractConnection  $connection
-         */
-        $shutdown = function ($channel, $connection) {
-            $channel->close();
-            $connection->close();
         };
 
-        register_shutdown_function($shutdown, $channel, $connection);
+        $channel->basic_qos(null, 1, false);
+        // no_ack = false：手动 ack
+        $channel->basic_consume($queue, $consumerTag, false, false, false, false, $process_message);
 
-        $timeout = 15;
-        // Loop as long as the channel has callbacks registered
-        while ($channel->is_consuming()) {
-            try {
-                $channel->wait(null, false, $timeout);
-            } catch (AMQPTimeoutException $e) {
-                // ignore it
-            }
+        try {
+            $channel->consume();
+        } catch (\Throwable $exception) {
+            Log::error($exception->getMessage());
         }
+
+        $channel->close();
+        $connection->close();
+
+        Log::info("mq worker stopped queue={$queue} processed={$processed}");
     }
 }

api-v13/app/Http/Controllers/SentenceController.php

@@ -516,7 +516,8 @@ class SentenceController extends Controller
             'para' => $param[1],
             'channel' => $channelId,
         ]);
-        Mq::publish('content', new SentResource($sent));
+        // FIXME: content 队列没有任何消费者，消息只会堆积，先停发。content 已从 mint.rabbitmq.queues 移除，确认无用后删掉此行
+        // Mq::publish('content', new SentResource($sent));
 
         if ($channel->type === 'nissaya' && $sent->content_type === 'json') {
             $this->updateWbwAnalyses($sent->content, $channel->lang, $user['user_id']);

api-v13/app/Services/RabbitMQService.php

@@ -2,7 +2,7 @@
 
 namespace App\Services;
 
-use App\Exceptions\TaskFailException;
+use App\Tools\Tools;
 use Illuminate\Support\Facades\Log;
 use Illuminate\Support\Str;
 use PhpAmqpLib\Channel\AMQPChannel;
@@ -10,41 +10,52 @@ use PhpAmqpLib\Connection\AMQPStreamConnection;
 use PhpAmqpLib\Message\AMQPMessage;
 use PhpAmqpLib\Wire\AMQPTable;
 
+/**
+ * RabbitMQ 的发布与消费。
+ *
+ * 消费策略：阻塞等待消息，每次只取一条；任务处理结束无论成功失败一律 ack，
+ * 不重试、不进死信队列。失败只记日志。
+ */
 class RabbitMQService
 {
-    private $connection;
+    private AMQPStreamConnection $connection;
 
-    private $channel;
+    private AMQPChannel $channel;
 
+    /**
+     * @var array{host: string, port: int|string, user: string, password: string, virtual_host: string}
+     */
     private array $config;
 
+    /**
+     * @var array<string, array{max_length?: int}>
+     */
     private array $queues;
 
     public function __construct()
     {
         $this->config = config('queue.connections.rabbitmq');
-        $this->queues = config('mint.rabbitmq.queues');
+        $this->queues = config('mint.rabbitmq.queues', []);
         $this->connect();
     }
 
-    private function connect()
+    private function connect(): void
     {
-        $conn = $this->config;
         $this->connection = new AMQPStreamConnection(
-            $conn['host'],
-            $conn['port'],
-            $conn['user'],
-            $conn['password'],
-            $conn['virtual_host']
+            $this->config['host'],
+            $this->config['port'],
+            $this->config['user'],
+            $this->config['password'],
+            $this->config['virtual_host']
         );
 
         $this->channel = $this->connection->channel();
 
-        // 设置 QoS - 每次只处理一条消息
+        // QoS：每次只投递一条未确认消息
         $this->channel->basic_qos(null, 1, null);
     }
 
-    public function isConnected()
+    public function isConnected(): bool
     {
         return $this->connection->isConnected();
     }
@@ -54,102 +65,32 @@ class RabbitMQService
         return $this->channel;
     }
 
+    /**
+     * 声明配置中的全部队列。
+     *
+     * @return list<string> 队列名
+     */
     public function createQueue(): array
     {
-        foreach ($this->queues as $name => $queue) {
-
-            $args = [];
-
-            // TTL
-            if (! empty($queue['ttl'])) {
-                $args['x-message-ttl'] = (int) $queue['ttl'];
-            }
-
-            // max length
-            if (! empty($queue['max_length'])) {
-                $args['x-max-length'] = (int) $queue['max_length'];
-            }
-
-            // dead letter exchange
-            if (! empty($queue['dead_letter_exchange'])) {
-                $args['x-dead-letter-exchange'] = $queue['dead_letter_exchange'];
-            }
-
-            // dead letter routing key（可选但建议）
-            if (! empty($queue['dead_letter_queue'])) {
-                $args['x-dead-letter-routing-key'] = $queue['dead_letter_queue'];
-            }
-
-            $this->channel->queue_declare(
-                $name,                 // queue 名称
-                false,                 // passive：检查是否存在（false=不存在则创建）
-                true,                  // durable：是否持久化（重启 RabbitMQ 后仍存在）
-                false,                 // exclusive：独占模式
-                false,                 // auto_delete：是否在无消费者时自动删除
-                false,                 // nowait：是否不等待服务器响应（false=阻塞等待确认）
-                new AMQPTable($args)   // arguments：附加参数（TTL / DLX / max-length 等）
-            );
+        foreach (array_keys($this->queues) as $name) {
+            $this->setupQueue($name);
         }
 
         return array_keys($this->queues);
     }
 
+    /**
+     * 声明一个持久化队列（不存在则创建）。
+     */
     public function setupQueue(string $queueName): void
     {
-        $queueConfig = config("mint.rabbitmq.queues.{$queueName}");
+        $queueConfig = $this->queues[$queueName] ?? [];
 
-        $workerArgs = [];
-        if (isset($queueConfig['ttl'])) {
-            $workerArgs['x-message-ttl'] = $queueConfig['ttl'];
-        }
-        if (isset($queueConfig['max_length'])) {
-            $workerArgs['x-max-length'] = $queueConfig['max_length'];
+        $args = [];
+        if (! empty($queueConfig['max_length'])) {
+            $args['x-max-length'] = (int) $queueConfig['max_length'];
         }
 
-        // 创建死信交换机
-        if (isset($queueConfig['dead_letter_exchange'])) {
-            $this->channel->exchange_declare(
-                $queueConfig['dead_letter_exchange'],
-                'direct',
-                false,
-                true,
-                false
-            );
-
-            $dlqName = $queueConfig['dead_letter_queue'];
-            $dlqConfig = config("mint.rabbitmq.dead_letter_queues.{$dlqName}", []);
-            $dlqArgs = [];
-            if (isset($dlqConfig['ttl'])) {
-                $dlqArgs['x-message-ttl'] = $dlqConfig['ttl'];
-            }
-            if (isset($dlqConfig['max_length'])) {
-                $dlqArgs['x-max-length'] = $dlqConfig['max_length'];
-            }
-            $dlqArguments = new AMQPTable($dlqArgs);
-
-            // 创建死信队列
-            $this->channel->queue_declare(
-                $dlqName,
-                false,  // passive
-                true,   // durable
-                false,  // exclusive
-                false,  // auto_delete
-                false,  // nowait
-                $dlqArguments
-            );
-
-            // 绑定死信队列到死信交换机
-            $this->channel->queue_bind(
-                $queueConfig['dead_letter_queue'],
-                $queueConfig['dead_letter_exchange']
-            );
-
-            // 主队列，配置死信
-            $workerArgs['x-dead-letter-exchange'] = $queueConfig['dead_letter_exchange'];
-            $workerArgs['x-dead-letter-routing-key'] = $queueConfig['dead_letter_queue'];
-        }
-        $arguments = new AMQPTable($workerArgs);
-
         $this->channel->queue_declare(
             $queueName,
             false,  // passive
@@ -157,47 +98,15 @@ class RabbitMQService
             false,  // exclusive
             false,  // auto_delete
             false,  // nowait
-            $arguments
+            new AMQPTable($args)
         );
-        /*
-        // 检查队列是否存在
-        try {
-            $this->channel->queue_declare(
-                $queueName,
-                true, // passive
-                true,   // durable
-                false, // exclusive
-                false, // auto_delete
-                false,  // nowait
-                $arguments
-            );
-            //存在，不用创建
-        } catch (\Exception $e) {
-            if (strpos($e->getMessage(), 'NOT_FOUND') !== false) {
-                Log::info("Queue $queueName does not exist.");
-                // 队列不存在，创建新队列
-                $this->channel->queue_declare(
-                    $queueName,
-                    false,  // passive
-                    true,   // durable
-                    false,  // exclusive
-                    false,  // auto_delete
-                    false,  // nowait
-                    $arguments
-                );
-            } else {
-                Log::error("Error checking queue $queueName: {$e->getMessage()}");
-                throw $e;
-            }
-        }
-            */
     }
 
     public function publishMessage(string $queueName, array $data): string
     {
         try {
             $this->setupQueue($queueName);
-            $msgId = Str::uuid();
+            $msgId = (string) Str::uuid();
             $message = new AMQPMessage(
                 json_encode($data, JSON_UNESCAPED_UNICODE),
                 [
@@ -221,123 +130,69 @@ class RabbitMQService
         }
     }
 
-    public function consume(string $queueName, callable $callback, ?int $maxIterations = null): void
+    /**
+     * 阻塞消费队列。
+     *
+     * 每条消息处理结束（成功、抛异常、JSON 解析失败）都 ack，不重试。
+     * 处理完 $loopLimit 条或检测到 .stop 文件后退出；$loopLimit 为 0 表示不限。
+     *
+     * @param  callable(mixed $data, AMQPMessage $message): void  $callback
+     * @param  int|null  $loopLimit  默认取 mint.mq.consumer_loop_limit
+     */
+    public function consume(string $queueName, callable $callback, ?int $loopLimit = null): void
     {
         $this->setupQueue($queueName);
-        $maxIterations = $maxIterations ?? $this->config['consumer']['max_iterations'];
-        $iteration = 0;
+        $loopLimit ??= (int) config('mint.mq.consumer_loop_limit', 0);
+        $processed = 0;
 
-        $consumerCallback = function (AMQPMessage $msg) use ($callback, $queueName, &$iteration) {
-            try {
-                $data = json_decode($msg->getBody(), true);
-                $retryCount = $this->getRetryCount($msg);
-                $maxRetries = $this->config['queues'][$queueName]['retry_count'];
+        $consumerCallback = function (AMQPMessage $msg) use ($callback, $queueName, $loopLimit, &$processed): void {
+            $messageId = $msg->has('message_id') ? $msg->get('message_id') : null;
+            $context = ['queue' => $queueName, 'message_id' => $messageId];
 
-                Log::info("Processing message from queue: {$queueName}", [
-                    'data' => $data,
-                    'retry_count' => $retryCount,
-                    'delivery_tag' => $msg->getDeliveryTag(),
-                ]);
-
-                // 执行回调处理消息
-                $callback($data, $retryCount);
-                // 处理成功，确认消息
-                $msg->ack();
-                Log::info('Message processed successfully', ['delivery_tag' => $msg->getDeliveryTag()]);
-            } catch (TaskFailException $e) {
-                // no need requeue
-                Log::error('Error processing message', [
+            try {
+                $data = json_decode($msg->getBody(), true, flags: JSON_THROW_ON_ERROR);
+                Log::info('mq message start', $context);
+                $callback($data, $msg);
+                Log::info('mq message done', $context);
+            } catch (\Throwable $e) {
+                Log::error('mq message failed', $context + [
                     'error' => $e->getMessage(),
-                    'delivery_tag' => $msg->getDeliveryTag(),
+                    'exception' => $e::class,
                 ]);
-                $msg->nack(false, false);
-            } catch (\Exception $e) {
-                // 处理失败，检查重试次数
-                if ($retryCount < $maxRetries) {
-                    // 重新入队，延迟处理
-                    $this->requeueWithDelay($msg, $queueName, $retryCount + 1);
-                    Log::warning('Message requeued for retry', [
-                        'delivery_tag' => $msg->getDeliveryTag(),
-                        'retry_count' => $retryCount + 1,
-                    ]);
-                } else {
-                    // 超过重试次数，拒绝消息（进入死信队列）
-                    $msg->nack(false, false);
-                    Log::error('Message rejected after max retries', [
-                        'delivery_tag' => $msg->getDeliveryTag(),
-                        'retry_count' => $retryCount,
-                    ]);
-                }
+            } finally {
+                $msg->ack();
             }
 
-            $iteration++;
+            $processed++;
+
+            if (($loopLimit > 0 && $processed >= $loopLimit) || Tools::isStop()) {
+                Log::info('mq consumer exit', $context + ['processed' => $processed]);
+                $msg->getChannel()->basic_cancel($msg->getConsumerTag());
+            }
         };
 
         $this->channel->basic_qos(null, 1, null);
-        $this->channel->basic_consume($queueName, '', false, false, false, false, $consumerCallback);
-
-        Log::info("Starting consumer for queue: {$queueName}", ['max_iterations' => $maxIterations]);
-
-        while ($this->channel->is_consuming() && $iteration < $maxIterations) {
-            $this->channel->wait(null, false, $this->config['consumer']['sleep_between_iterations']);
-        }
-
-        Log::info('Consumer stopped', ['iterations_processed' => $iteration]);
-    }
-
-    private function getRetryCount(AMQPMessage $msg): int
-    {
-        $headers = $msg->get_properties();
-
-        return isset($headers['application_headers']['x-retry-count'])
-            ? $headers['application_headers']['x-retry-count'] : 0;
-    }
-
-    private function requeueWithDelay(AMQPMessage $msg, string $queueName, int $retryCount): void
-    {
-        $delay = $this->config['queues'][$queueName]['retry_delay'];
-
-        // 创建延迟队列
-        $delayQueueName = "{$queueName}_delay_{$retryCount}";
-        $arguments = new AMQPTable([
-            'x-message-ttl' => $delay,
-            'x-dead-letter-exchange' => '',
-            'x-dead-letter-routing-key' => $queueName,
-        ]);
-
-        $this->channel->queue_declare(
-            $delayQueueName,
-            false,
-            true,
-            false,
-            false,
-            false,
-            $arguments
+        $this->channel->basic_consume(
+            $queueName,
+            '',     // consumer_tag
+            false,  // no_local
+            false,  // no_ack：手动 ack
+            false,  // exclusive
+            false,  // nowait
+            $consumerCallback
         );
 
-        // 重新发布消息到延迟队列
-        $data = json_decode($msg->getBody(), true);
-        $newMessage = new AMQPMessage(
-            json_encode($data),
-            [
-                'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
-                'application_headers' => new AMQPTable([
-                    'x-retry-count' => $retryCount,
-                ]),
-            ]
-        );
+        Log::info("mq consumer start queue={$queueName}", ['loop_limit' => $loopLimit]);
 
-        $this->channel->basic_publish($newMessage, '', $delayQueueName);
-        $msg->ack();
+        // 阻塞等待，直到回调里 basic_cancel
+        $this->channel->consume();
+
+        Log::info("mq consumer stopped queue={$queueName}", ['processed' => $processed]);
     }
 
     public function close(): void
     {
-        if ($this->channel) {
-            $this->channel->close();
-        }
-        if ($this->connection) {
-            $this->connection->close();
-        }
+        $this->channel->close();
+        $this->connection->close();
     }
 }

api-v13/config/mint.php

@@ -150,39 +150,18 @@ return [
         ],
     ],
     'mq' => [
-        'loop_limit' => [
-            'ai_translate' => env('MQ_LOOP_LIMIT_AI_TRANSLATE', 0),
-        ],
+        'consumer_loop_limit' => env('MQ_CONSUMER_LOOP_LIMIT', 0),
     ],
     'rabbitmq' => [
         'queues' => [
-            'task' => ['retry_times' => env('RABBITMQ_AI_RETRY_TIMES', 3)],
-            'discussion' => ['retry_times' => env('RABBITMQ_AI_RETRY_TIMES', 3)],
-            'export_pali_chapter' => ['retry_times' => env('RABBITMQ_AI_RETRY_TIMES', 3)],
-            'export_article' => ['retry_times' => env('RABBITMQ_AI_RETRY_TIMES', 3)],
-            'progress' => ['retry_times' => env('RABBITMQ_AI_RETRY_TIMES', 3)],
-            'content' => ['retry_times' => env('RABBITMQ_AI_RETRY_TIMES', 3)],
-            'suggestion' => ['retry_times' => env('RABBITMQ_AI_RETRY_TIMES', 3)],
-            'wbw-analyses' => ['retry_times' => env('RABBITMQ_AI_RETRY_TIMES', 3)],
-            'ai_translate_v2' => [
-                'retry_times' => env('RABBITMQ_AI_RETRY_TIMES', 3),
-                'max_loop_count' => env('RABBITMQ_AI_MAX_LOOP', 10),
-                'timeout' => env('RABBITMQ_AI_TIMEOUT', 300),
-                'dead_letter_queue' => 'ai_translate_dlq',
-                'dead_letter_exchange' => 'ai_translate_dlx',
-            ],
-            'heartbeat_queue' => [
-                'ttl' => 86400000, // 24小时 TTL (毫秒)
-                'max_length' => 10000,
-            ],
-        ],
-
-        // 死信队列配置
-        'dead_letter_queues' => [
-            'ai_translate_dlq' => [
-                'ttl' => 86400000, // 24小时 TTL (毫秒)
-                'max_length' => 10000,
-            ],
+            'task' => [],
+            'discussion' => [],
+            'export_pali_chapter' => [],
+            'export_article' => [],
+            'progress' => [],
+            'suggestion' => [],
+            'wbw-analyses' => [],
+            'ai_translate_v2' => [],
         ],
     ],
     'opensearch' => [

api-v13/documents/tasks.md

@@ -4,21 +4,41 @@
 
 ## MQ workers
 
+worker 不再自建队列，队列不存在会报错退出。部署或清空 RabbitMQ 后先建队列（队列列表见 `config/mint.php` 的 `mint.rabbitmq.queues`）：
+
 ```bash
+php artisan app:create-queue
+```
+
+每条消息处理完无论成功失败都 ack，不重试、不进死信队列。
+处理满 `MQ_CONSUMER_LOOP_LIMIT` 条（0 = 不限）或检测到 `.stop` 文件后 worker 退出，需要外部重新拉起。
+
+```bash
+# 队列 discussion
 php artisan mq:discussion
 
+# 队列 export_pali_chapter
 php artisan mq:export.pali.chapter
 
+# 队列 export_article
 php artisan mq:export.article
 
+# 队列 progress
 php artisan mq:progress
 
+# 队列 suggestion
+php artisan mq:pr
+
+# 队列 task
 php artisan mq:task
 
+# 队列 wbw-analyses
 php artisan mq:wbw.analyses
 
 ```
 
+`ai_translate_v2`（`rabbitmq:consume`）待第二阶段重构 worker 后补上。`content` 队列已移除（无消费者，发送端已注释，见 `SentenceController@update` 的 FIXME）。
+
 ## every five minutes
 
 ```bash