workHour/app/Jobs/ProcessOssTask.php

<?php

namespace App\Jobs;

use App\Model\SysOssTasks;
use App\Service\FileUploadService;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;
use Symfony\Component\Console\Output\ConsoleOutput;
use Symfony\Component\Console\Output\OutputInterface;

class ProcessOssTask implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    public $timeout = 300;

    public function __construct()
    {

    }

    public function handle()
    {
        $currentBatchIds = [];

        try {
            $service = new FileUploadService();
            SysOssTasks::where('status', 0)
                ->chunk(50, function ($tasks) use (&$currentBatchIds, $service) {
                    // 每次进入新的一批，先记录这一批的所有 ID
                    $currentBatchIds = $tasks->pluck('id')->toArray();

                    foreach ($tasks as $task) {
                        $status_db = 2;
                        if($task['type'] == 1){
                            //删除
                            list($status, $msg) = $service->createOssUploadOldSingle($task->url);
                            if($status) $status_db = 1;

                        }elseif ($task['type'] == 2){
                            //新增
                            list($status, $msg) = $service->createOssUploadSingle($task->url);
                            if($status) $status_db = 1;
                        }else{
                            $msg = "单条失败：类型不存在";
                        }

                        $task->update([
                            'status' => $status_db,
                            'remark' => $msg,
                            'upd_time' => time()
                        ]);
                    }

                    // 处理完这一批后，清空当前批次 ID 记录
                    $currentBatchIds = [];
                });
        } catch (\Throwable $e) {
            $errorMessage = "当前批次执行中断：" . $e->getMessage();
            if (!empty($currentBatchIds)) {
                SysOssTasks::whereIn('id', $currentBatchIds)
                    ->where('status', 0)
                    ->update([
                        'status' => 2,
                        'remark' => $errorMessage,
                        'upd_time' => time(),
                    ]);
            }

            $this->delete();
        }
    }

    protected function echoMessage(OutputInterface $output)
    {
        //输出消息
        $output->writeln(json_encode($this->data));
    }
}