yusheng-php/application/common/command/ProcessGiftQueue.php

<?php
// application/common/command/ProcessGiftQueue.php

namespace app\common\command;

use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\console\input\Argument;
use think\console\input\Option;
use think\Log;
use app\common\library\GiftQueue;

class ProcessGiftQueue extends Command
{
    protected function configure()
    {
        $this->setName('process:gift_queue')
            ->setDescription('处理送礼队列')
            ->addArgument('mode', Argument::OPTIONAL, '运行模式: once, daemon', 'once')
            ->addOption('batch-size', 'b', Option::VALUE_OPTIONAL, '每次处理数量', 100)
            ->addOption('sleep-time', 's', Option::VALUE_OPTIONAL, '处理间隔(毫秒)', 100)
            ->addOption('max-runtime', 't', Option::VALUE_OPTIONAL, '最大运行时间(秒)', 0);
    }

    protected function execute(Input $input, Output $output)
    {
        $mode = $input->getArgument('mode');
        $batchSize = $input->getOption('batch-size');
        $sleepTime = $input->getOption('sleep-time');
        $maxRuntime = $input->getOption('max-runtime');

        $output->writeln("开始处理送礼队列...");
        $output->writeln("模式: {$mode}, 批量大小: {$batchSize}, 间隔: {$sleepTime}ms");

        $startTime = time();
        $totalProcessed = 0;
        $totalSuccess = 0;
        $totalFailed = 0;

        // 检查Redis连接
        try {
            $queueSize = GiftQueue::size();
            $output->writeln("当前队列大小: {$queueSize}");
        } catch (\Exception $e) {
            $output->writeln("<error>Redis连接失败: " . $e->getMessage() . "</error>");
            return;
        }

        do {
            $queueSize = GiftQueue::size();

            if ($queueSize == 0) {
                if ($mode == 'once') {
                    $output->writeln("队列为空，处理完成");
                    break;
                } else {
                    $output->writeln("队列为空，等待新数据...");
                    sleep(5);
                    continue;
                }
            }

            $output->writeln("当前队列大小: {$queueSize}");

            // 处理队列
            $result = GiftQueue::process($batchSize);

            $totalProcessed += $result['processed'];
            $totalSuccess += $result['success'];
            $totalFailed += $result['failed'];

            $output->writeln(sprintf(
                "处理完成：已处理 %d，成功 %d，失败 %d",
                $result['processed'],
                $result['success'],
                $result['failed']
            ));

            // 休眠
            usleep($sleepTime * 1000);

            // 检查运行时间限制
            if ($maxRuntime > 0 && (time() - $startTime) > $maxRuntime) {
                $output->writeln("达到最大运行时间，停止处理");
                break;
            }

            // 单次模式处理完成后退出
            if ($mode == 'once' && $result['processed'] < $batchSize) {
                $output->writeln("队列处理完成");
                break;
            }

        } while ($mode == 'daemon');

        $output->writeln("队列处理统计:");
        $output->writeln("总计处理: {$totalProcessed}");
        $output->writeln("成功: {$totalSuccess}");
        $output->writeln("失败: {$totalFailed}");
        $output->writeln("剩余队列大小: " . GiftQueue::size());

        Log::info(sprintf(
            "送礼队列处理统计: 总计 %d, 成功 %d, 失败 %d, 剩余 %d",
            $totalProcessed,
            $totalSuccess,
            $totalFailed,
            GiftQueue::size()
        ));
    }
}