如何利用Bus Chain拆分长耗时的VAT验证Cron任务

核心思路 #

将原有的单步VAT验证任务拆分为两个独立子任务，通过任务链串联执行：

1. 数据预处理暂存：分批拉取并清洗采购订单数据，存入临时表，避免单次加载大量数据
2. 税率差异对比记录：基于临时表的批次数据，按订单分组检查税率差异并记录日志

1. 实现数据预处理任务 #

创建独立的预处理任务类，负责批量处理原始数据并暂存，同时触发后续对比任务：

class PurchaseVatDataPrepareTask extends AbstractTask
{
    public function execute()
    {
        // 生成唯一批次ID，用于关联后续任务的数据
        $batchId = uniqid('vat_batch_', true);
        
        // 分批拉取未处理的采购订单，避免一次性加载过多数据
        PurchaseOrder::query()
            ->where('vat_validated', 0)
            ->chunk(100, function ($orderChunk) use ($batchId) {
                $tempRecords = [];
                foreach ($orderChunk as $order) {
                    $tempRecords[] = [
                        'batch_id' => $batchId,
                        'order_id' => $order->id,
                        'item_id' => $order->item_id,
                        'vat_rate' => $order->vat_rate,
                        'amount' => $order->amount,
                        'created_at' => now()
                    ];
                }
                
                // 批量插入临时表，提升写入效率
                DB::table('temp_purchase_vat_data')->insert($tempRecords);
                
                // 标记这批订单为待验证状态，避免重复处理
                PurchaseOrder::whereIn('id', collect($orderChunk)->pluck('id'))
                    ->update(['vat_validated' => 1]);
            });
        
        // 触发税率对比任务，传递批次ID
        $this->dispatch(new PurchaseVatDiffCheckTask($batchId));
    }
}

2. 实现税率差异对比任务 #

创建对比任务类，基于批次ID从临时表取数，检查同一订单内的税率差异并记录：

class PurchaseVatDiffCheckTask extends AbstractTask
{
    protected $batchId;

    public function __construct(string $batchId)
    {
        $this->batchId = $batchId;
    }

    public function execute()
    {
        // 按订单分组，统计同一订单内的不同税率数量
        $diffOrders = DB::table('temp_purchase_vat_data')
            ->where('batch_id', $this->batchId)
            ->select(
                'order_id',
                DB::raw('COUNT(DISTINCT vat_rate) as diff_count'),
                DB::raw('GROUP_CONCAT(DISTINCT vat_rate SEPARATOR ", ") as vat_rates')
            )
            ->groupBy('order_id')
            ->having('diff_count', '>', 1)
            ->get();

        // 批量记录税率差异日志
        $logEntries = [];
        foreach ($diffOrders as $order) {
            $logEntries[] = [
                'order_id' => $order->order_id,
                'vat_rates' => $order->vat_rates,
                'diff_count' => $order->diff_count,
                'created_at' => now()
            ];
        }
        
        if (!empty($logEntries)) {
            DB::table('purchase_vat_diff_logs')->insert($logEntries);
        }

        // 清理当前批次的临时数据（可选，根据数据保留需求调整）
        DB::table('temp_purchase_vat_data')->where('batch_id', $this->batchId)->delete();
    }
}

3. 调整Cron任务配置 #

将原有的Cron任务改为触发预处理任务，替换原有的purchaseVatValidationCron逻辑：

protected function schedule(Schedule $schedule)
{
    // 按业务需求调整执行频率，比如每日执行
    $schedule->call(function () {
        dispatch(new PurchaseVatDataPrepareTask());
    })->daily();
}

关键注意事项 #

• 临时表优化：给temp_purchase_vat_data的batch_id和order_id字段添加索引，提升分组查询效率
• 任务重试机制：在任务类中添加protected $tries = 3;，避免因临时网络或数据库问题导致任务失败
• 状态字段设计：确保PurchaseOrder表的vat_validated字段（0=未处理，1=待验证，2=已完成）能准确标记订单处理状态
• 分批粒度调整：根据服务器性能调整chunk方法的参数（示例中为100），平衡内存占用和处理效率

内容的提问来源于stack exchange，提问作者user22809416