Laravel任务批处理？批量任务如何使用？

IlluminateBusBatch

Bus

IlluminateBusBatchable

<?php  namespace appJobs;  use IlluminateBusBatchable; // 引入Batchable trait use IlluminateBusQueueable; use IlluminateContractsQueueShouldQueue; use IlluminateFoundationBusDispatchable; use IlluminateQueueInteractsWithQueue; use IlluminateQueueSerializesModels;  class ProcessCsvRow implements ShouldQueue {     use Dispatchable, InteractsWithQueue, Queueable, SerializesModels, Batchable; // 使用Batchable      public $data;      public function __construct(array $data)     {         $this->data = $data;     }      public function handle(): void     {         // 模拟处理一行CSV数据，可能涉及数据库操作或外部API调用         sleep(1); // 模拟耗时操作         Log::info('Processing data: ' . JSon_encode($this->data));         // 假设这里可能会有业务逻辑错误         if (rand(0, 10) < 1) { // 10%的几率失败             throw new Exception('Failed to process data for: ' . json_encode($this->data));         }     } }

<?php  namespace ApphttpControllers;  use AppJobsProcessCsvRow; use IlluminateHttpRequest; use IlluminateSupportFacadesBus; use Throwable;  class CsvUploadController extends Controller {     public function upload(Request $request)     {         $file = $request->file('csv_file');         $rows = array_map('str_getcsv', file($file->getRealPath()));         $header = array_shift($rows); // 假设第一行是表头          $jobs = [];         foreach ($rows as $row) {             $jobs[] = new ProcessCsvRow(array_combine($header, $row));         }          $batch = Bus::batch($jobs)             ->then(function (IlluminateBusBatch $batch) {                 // 所有任务成功完成时执行                 Log::info("Batch {$batch->id} completed successfully.");                 // 比如发送成功通知邮件             })             ->catch(function (IlluminateBusBatch $batch, Throwable $e) {                 // 批次中的任何任务失败时执行                 Log::error("Batch {$batch->id} failed: " . $e->getMessage());                 // 比如发送失败告警，并记录失败的任务             })             ->finally(function (IlluminateBusBatch $batch) {                 // 批次完成（无论成功或失败）时执行                 Log::info("Batch {$batch->id} finished processing (finally).");                 // 清理临时文件，更新ui状态等             })             ->name('CSV Import ' . now()->format('YmdHis')) // 给批次命名，方便追踪             ->onConnection('redis') // 指定队列连接             ->onQueue('imports') // 指定队列名称             ->dispatch(); // 分发批次          return response()->json([             'message' => 'CSV file uploaded and processing started.',             'batch_id' => $batch->id,         ]);     } }

Bus::batch($jobs)

then()

catch()

finally()

dispatch()

name()

onConnection()

onQueue()

搜狐资讯

AI资讯助手，追踪所有你关心的信息

24

查看详情

// 幂等性示例：在Job中检查数据是否存在 public function handle(): void {     $identifier = $this->data['unique_id'] ?? null;     if ($identifier && AppModelsProcessedItem::where('identifier', $identifier)->exists()) {         Log::info("Item with identifier {$identifier} already processed. Skipping.");         return;     }      // ... 实际处理逻辑 ...      AppModelsProcessedItem::create(['identifier' => $identifier, 'status' => 'completed']); }

Batch

use IlluminateSupportFacadesBus;  // 假设你从前端或其他地方获取到了batchId $batchId = 'some-uuid-from-job-batches-table'; $batch = Bus::findBatch($batchId);  if ($batch) {     echo "Batch Name: " . $batch->name . "n";     echo "Total Jobs: " . $batch->totalJobs . "n";     echo "Pending Jobs: " . $batch->pendingJobs . "n";     echo "Processed Jobs: " . $batch->processedJobs . "n";     echo "Failed Jobs: " . $batch->failedJobs . "n";     echo "Progress: " . $batch->progress() . "%n";     echo "Finished: " . ($batch->finished() ? 'Yes' : 'No') . "n";     echo "Cancelled: " . ($batch->cancelled() ? 'Yes' : 'No') . "n";     echo "Failed: " . ($batch->failed() ? 'Yes' : 'No') . "n";      // 甚至可以获取失败的Job ID     foreach ($batch->failedJobs() as $failedJobId) {         // 根据failedJobId进一步查询失败原因         // 注意：这里返回的是Job ID，不是Batch ID         // 你可能需要从queue_failures表或其他地方查找详细信息         echo "Failed Job ID: " . $failedJobId . "n";     } } else {     echo "Batch not found.n"; }

AppProvidersEventServiceProvider

// App/Providers/EventServiceProvider.php protected $listen = [     IlluminateBusEventsBatchFinished::class => [         AppListenersLogBatchFinished::class,     ],     IlluminateBusEventsBatchFailed::class => [         AppListenersNotifyBatchFailed::class,     ],     // ... ];

// App/Listeners/NotifyBatchFailed.php <?php  namespace AppListeners;  use IlluminateBusEventsBatchFailed; use IlluminateContractsQueueShouldQueue; use IlluminateQueueInteractsWithQueue; use IlluminateSupportFacadesLog; use IlluminateSupportFacadesMail; use AppMailBatchFailureNotification; // 假设你有一个邮件通知类  class NotifyBatchFailed {     public function handle(BatchFailed $event): void     {         $batch = $event->batch;         Log::error("Batch {$batch->id} named '{$batch->name}' failed. Total jobs: {$batch->totalJobs}, Failed jobs: {$batch->failedJobs}.");          // 发送邮件通知管理员         // Mail::to('admin@example.com')->send(new BatchFailureNotification($batch));          // 记录更详细的失败Job信息         foreach ($batch->failedJobs() as $failedJobId) {             Log::error("Failed job ID in batch {$batch->id}: {$failedJobId}");             // 这里可以进一步查询 queue_failures 表来获取异常堆栈         }     } }