mp_audio_manage/app/Console/Commands/ProcessBatchEpisodeGenerationCommand.php

<?php

namespace App\Console\Commands;

use Illuminate\Console\Command;
use Illuminate\Support\Facades\DB;
use Symfony\Component\Process\Process;

/**
 * 批量生成分集定时任务（调度器）
 * 每次执行：找出有 pending 任务的 anime_id，为最多 N 个（默认5）空闲 anime 各拉起一个后台子进程，
 * 由子进程独占处理对应 anime 的待生成剧集。
 */
class ProcessBatchEpisodeGenerationCommand extends Command
{
    /**
     * The name and signature of the console command.
     *
     * @var string
     */
    protected $signature = 'batch:generate-episodes';

    /**
     * The console command description.
     *
     * @var string
     */
    protected $description = '批量生成分集定时任务：并发调度不同anime_id的子进程处理，并发数默认5';

    /**
     * Execute the console command.
     *
     * @return int
     */
    public function handle()
    {
        dLog('command')->info('开始执行批量生成分集调度任务...');

        $concurrency = (int) env('BATCH_EPISODE_GENERATION_CONCURRENCY', 5);
        if ($concurrency < 1) {
            $concurrency = 1;
        }

        // 找出所有有待处理任务的 anime_id
        $animeIds = DB::table('mp_batch_episode_generation_details')
            ->where('status', 'pending')
            ->distinct()
            ->pluck('anime_id')
            ->map(function ($animeId) {
                return (int) $animeId;
            })
            ->sort()
            ->values()
            ->toArray();

        if (empty($animeIds)) {
            dLog('command')->info('没有待处理的任务，本次调度结束');
            return 0;
        }

        $spawned = 0;
        foreach ($animeIds as $animeId) {
            if ($spawned >= $concurrency) {
                dLog('command')->info("已达到最大并发数 {$concurrency}，停止调度");
                break;
            }

            $lockName = ProcessBatchAnimeEpisodeGenerationCommand::LOCK_PREFIX . $animeId;

            // 仅当该 anime 的处理锁空闲时才拉起子进程；
            // 真正的加锁由子进程完成，避免锁挂在调度进程的连接上
            $isFree = DB::selectOne('SELECT IS_FREE_LOCK(?) AS is_free', [$lockName]);
            if (!$isFree || (int) $isFree->is_free !== 1) {
                dLog('command')->info("anime_id: {$animeId} 正在被其他进程处理，跳过");
                continue;
            }

            try {
                $this->spawnWorker($animeId);
                $spawned++;
                dLog('command')->info("已拉起子进程处理 anime_id: {$animeId}");
            } catch (\Throwable $e) {
                dLog('command')->error("拉起子进程失败 anime_id: {$animeId}: " . $e->getMessage());
                logDB('command', 'error', '批量分集调度：拉起子进程失败', [
                    'anime_id' => $animeId,
                    'error' => $e->getMessage(),
                    'trace' => $e->getTraceAsString()
                ]);
            }
        }

        dLog('command')->info("批量生成分集调度完成，本次拉起子进程数: {$spawned}");
        return 0;
    }

    /**
     * 拉起一个后台子进程，独占处理指定 anime_id 的待生成剧集
     *
     * @param int $animeId
     * @return void
     */
    private function spawnWorker($animeId)
    {
        $process = new Process([
            PHP_BINARY,
            base_path('artisan'),
            'batch:generate-episodes:anime',
            (string) $animeId,
        ]);
        $process->setTimeout(null);
        $process->setIdleTimeout(null);
        $process->disableOutput();
        $process->start();
    }
}