ThinkPHP框架中数据库操作之：ThinkORM 大数据处理全攻略：告别内存溢出，轻松应对海量数据

在现代Web应用开发中，处理大数据集一直是一个挑战。当数据量达到百万甚至千万级别时，传统的查询方式往往会导致内存溢出，使应用崩溃。ThinkORM 4.0 针对这一问题提供了一系列高效的大数据处理方法，让开发者能够轻松应对海量数据的查询和处理。

大数据处理的挑战

传统的 select() 方法会将所有查询结果一次性加载到内存中，当数据量庞大时，这种方式会迅速消耗可用内存，最终导致内存溢出错误。想象一下，当你的应用需要导出一张百万级用户表或处理大量日志数据时，传统方法显然力不从心。

ThinkORM 4.0 提供了几种专门针对大数据集的解决方案，它们各有特点，适用于不同的场景：

cursor() - 游标查询，逐行返回数据
lazy() - 惰性加载，分批查询数据
stream() - 内部调用cursor方法实现并支持回调
chunk() - 内部调用lazy方法实现并支持回调

游标查询 (Cursor)

工作原理

游标查询采用逐行处理的方式，每次从数据库只取一条数据，处理完后再取下一条，直到所有数据处理完毕。这种方式使用PHP Generator实现，内存占用极低。

use think\facade\Db;

// 基础游标查询
foreach (Db::table('user')->cursor() as $user) {
    // 处理单条记录
    echo $user['name'] . PHP_EOL;
}

// 带条件的游标查询
$cursor = Db::table('order')
    ->where('status', 1)
    ->where('created_time', '>', '2024-01-01')
    ->cursor();
    
foreach ($cursor as $order) {
    // 逐行处理订单
    processOrder($order);
}

适用场景

游标查询特别适合以下场景：

导出大量数据到文件
数据清洗和转换
不需要随机访问数据的顺序处理

使用stream方法进行回调处理

如果需要使用回调方式处理数据，可以直接使用 stream 方法，它是 cursor 的封装：

// 返回处理的数据数量
$count = Db::table('user')->stream(function($user) {
    // 处理单条记录
    echo $user['name'] . PHP_EOL;
    // 返回 false 可以中断处理
    // return false;
});

echo "总共处理了 {$count} 条数据";

惰性查询 (Lazy)

工作原理

惰性查询采用分批处理的方式，例如先查询1-1000条数据，处理完后再查询1001-2000条，以此类推。这种方式相比游标查询有一些独特的优势。

// 默认每次查询1000条
foreach (Db::table('user')->lazy() as $user) {
    // 处理用户
}

// 使用自定义字段分批
foreach (Db::table('order')->lazy(500, 'order_no', 'asc') as $order) {
    // 处理订单
}

优势

支持关联查询（预加载）
支持链式操作（filter、map等）

使用chunk方法进行回调处理

chunk 方法内部调用了 lazy 方法并提供了回调机制，使数据处理更加方便：

Db::table('user')->chunk(500, function($users){
    foreach($users as $user) {
        // 处理用户数据
    }
});

需要注意的是，chunk 回调方法的参数是批次数据（数组），而不是单个数据。

实际应用场景

场景1：导出大量数据到CSV

// 使用 cursor - 内存占用最小
$file = fopen('export.csv', 'w');
fputcsv($file, ['id', 'name', 'email', 'register_time']);

Db::table('users')
    ->field('id,name,email,register_time')
    ->cursor(true) // 无缓冲查询
    ->each(function ($user) use ($file) {
        fputcsv($file, [
            $user['id'],
            $user['name'],
            $user['email'],
            $user['register_time']
        ]);
    });

fclose($file);

场景2：数据清洗和转换

// 使用 lazy - 支持链式操作
$processedCount = Db::table('raw_data')
    ->lazy(500)
    ->filter(function ($row) {
        // 过滤无效数据
        return !empty($row['email']) && filter_var($row['email'], FILTER_VALIDATE_EMAIL);
    })
    ->map(function ($row) {
        // 数据转换
        return [
            'email' => strtolower($row['email']),
            'name' => ucwords($row['name']),
            'processed_at' => date('Y-m-d H:i:s')
        ];
    })
    ->each(function ($data) {
        // 保存处理后的数据
        Db::table('clean_data')->insert($data);
    })
    ->count();

echo "Processed $processedCount records\n";

场景3：处理关联数据

// 使用 lazy - 支持关联查询
use app\model\User;

// 批量加载用户及其订单
User::with(['orders'])
    ->lazy(100)
    ->each(function ($user) {
        echo "User: {$user->name}\n";
        foreach ($user->orders as $order) {
            echo "  Order #{$order->id}: {$order->total}\n";
        }
    });

注意事项和最佳实践

缓存与流式查询

流式查询不适合与查询缓存一起使用，因为：

游标查询使用了Generator机制，无法序列化缓存
流式查询的目的是处理大数据集，缓存会失去意义

如果需要使用查询缓存，应该使用 select() 方法：

// 正确：使用 select 进行缓存
$users = Db::table('user')
    ->where('status', 1)
    ->cache(600)
    ->select();

// 错误：cursor 不支持缓存
$cursor = Db::table('user')
    ->cache(600) // 这个会被忽略
    ->cursor();

关联查询与流式处理

在使用模型的流式查询时，如果需要使用预载入关联查询，可以使用 lazy 方法：

User::with(['profile'])->lazy()
->each(function($user) {
    echo $user->profile->bio;
});

cursor 查询默认不支持关联，但可以调用 LazyCollection 数据集对象的 load 方法实现：

User::cursor()->load(['profile'])
->each(function($user) {
    echo $user->profile->bio;
});

无缓冲查询（MySQL专用）

对于超大数据集，MySQL数据库可以使用无缓冲查询模式：

// 在非MySQL数据库上，无缓冲查询模式选项会被忽略
$cursor = Db::table('big_table')->cursor(true);
foreach ($cursor as $row) {
    // 处理数据
    // 注意：无缓冲查询模式下，必须读取完所有结果才能执行下一个查询
}

中断处理

所有流式查询都支持中断：

// cursor/stream 返回 false 中断
Db::table('user')->stream(function($user) {
    if ($user['id'] == 100) {
        return false;  // 停止处理
    }
    // 处理数据
});

// chunk 返回 false 中断
Db::table('user')->chunk(100, function($users) {
    // 处理批次
    if (time() > $endTime) {
        return false;  // 超时停止
    }
});

性能优化建议

选择合适的方法：根据数据量大小和需求选择最合适的方法
- 简单导出 → cursor() / stream()
- 需要关联 → lazy()
- 批量操作 → chunk()
- 小数据量 → select()
合理设置批次大小：lazy() 和 chunk() 的批次大小一般在 500-2000 之间
避免在循环中执行查询：这会导致 N+1 查询问题
利用LazyCollection的强大功能：

Db::table('orders')
    ->lazy()
    ->filter(fn($order) => $order['amount'] > 100)
    ->map(fn($order) => [
        'amount' => $order['amount'] * 1.1  // 加价 10%
    ])
    ->take(1000)  // 只处理前 1000 条
    ->each(function ($chunk) {
        // 处理数据块
    });