PHP项目Laravel chunk分批处理数据

本文目录导读：

1. 基本用法
2. chunkById() - 推荐用于更新操作
3. 高级用法
4. 实际应用示例
5. 注意事项
6. 性能优化技巧
7. 监控处理进度

在Laravel中,chunk()方法用于分批处理大量数据，避免内存溢出，以下是详细的使用方法和最佳实践：

基本用法

chunk() 基础方法

use App\Models\User;
// 每次处理100条记录
User::chunk(100, function ($users) {
    foreach ($users as $user) {
        // 处理每条记录
        echo $user->name;
    }
});

带条件的chunk

// 只处理符合条件的记录
User::where('status', 'active')
    ->chunk(200, function ($users) {
        foreach ($users as $user) {
            // 处理逻辑
        }
    });

chunkById() - 推荐用于更新操作

当在chunk回调中更新数据时,使用chunkById()更安全：

use App\Models\User;
// 按ID排序处理，适合在回调中更新数据
User::chunkById(100, function ($users) {
    foreach ($users as $user) {
        User::where('id', $user->id)
            ->update(['status' => 'processed']);
    }
});

高级用法

带条件的chunkById

User::where('status', 'pending')
    ->orderBy('id')
    ->chunkById(100, function ($users) {
        foreach ($users as $user) {
            // 处理逻辑
        }
    }, 'id'); // 指定主键字段

使用游标（更高效的内存使用）

use App\Models\User;
// 使用游标逐条处理，内存占用更低
foreach (User::where('status', 'active')->cursor() as $user) {
    // 处理每个用户
    processUser($user);
}

实际应用示例

批量更新用户状态

use App\Models\User;
use Illuminate\Support\Facades\Log;
public function processUsers()
{
    $processed = 0;
    User::where('status', 'pending')
        ->chunkById(200, function ($users) use (&$processed) {
            foreach ($users as $user) {
                try {
                    // 模拟处理逻辑
                    // $this->someService($user);
                    $user->status = 'processed';
                    $user->save();
                    $processed++;
                } catch (\Exception $e) {
                    Log::error("处理用户失败: {$user->id}", [
                        'error' => $e->getMessage()
                    ]);
                }
            }
            Log::info("处理完成一批，当前处理: {$processed} 条");
        });
    return $processed;
}

大数据导出

public function exportData()
{
    $filename = 'export_' . date('Y-m-d_His') . '.csv';
    $file = fopen(storage_path('app/' . $filename), 'w');
    // 写入表头
    fputcsv($file, ['ID', '姓名', '邮箱', '创建时间']);
    User::chunk(500, function ($users) use ($file) {
        foreach ($users as $user) {
            fputcsv($file, [
                $user->id,
                $user->name,
                $user->email,
                $user->created_at
            ]);
        }
    });
    fclose($file);
    return $filename;
}

清理过期数据

use App\Models\Token;
public function cleanupExpiredTokens()
{
    $cutoffDate = now()->subDays(30);
    Token::where('created_at', '<', $cutoffDate)
        ->chunkById(500, function ($tokens) {
            foreach ($tokens as $token) {
                $token->delete();
            }
        });
}

注意事项

避免的问题

// ❌ 错误示范：在chunk中更新可能导致数据跳过
User::chunk(100, function ($users) {
    foreach ($users as $user) {
        User::where('id', $user->id)->delete();
    }
});
// ✅ 正确做法：使用chunkById
User::chunkById(100, function ($users) {
    foreach ($users as $user) {
        User::where('id', $user->id)->delete();
    }
});

与队列结合使用

use App\Jobs\ProcessUser;
use App\Models\User;
public function dispatchJobs()
{
    User::where('status', 'pending')
        ->chunk(100, function ($users) {
            foreach ($users as $user) {
                ProcessUser::dispatch($user->id);
            }
        });
}

性能优化技巧

// 只选择需要的字段
User::select('id', 'name', 'email')
    ->where('status', 'active')
    ->chunk(500, function ($users) {
        // 处理逻辑
    });
// 使用with预加载关联数据
User::with(['orders', 'profile'])
    ->chunk(200, function ($users) {
        foreach ($users as $user) {
            // $user->orders 和 $user->profile 已经加载
        }
    });

监控处理进度

public function processWithProgress()
{
    $total = User::where('status', 'pending')->count();
    $processed = 0;
    User::where('status', 'pending')
        ->chunkById(200, function ($users) use (&$processed, $total) {
            foreach ($users as $user) {
                // 处理逻辑
                $processed++;
            }
            $percent = round(($processed / $total) * 100, 2);
            Log::info("处理进度: {$percent}% ({$processed}/{$total})");
        });
}

• chunk(): 适用于只读操作或不会影响查询结果集的场景
• chunkById(): 推荐用于更新/删除操作
• cursor(): 需要逐条处理且内存要求极低时使用
• 始终在处理逻辑中包含错误处理
• 根据服务器资源调整每批处理的数量（通常100-500条为宜）