并发处理

什么是并发处理

对多个外部 API 的请求、数据库的聚合等相互不依赖的处理，如果顺序执行，总耗时会等于每个任务耗时的累加。如果把它们同时执行，那么总耗时就会缩短到最慢那一项的耗时。 Laravel 的 Concurrency Facade 通过简洁的 API 实现了这种并行执行。

Concurrency Facade 在 Laravel 11 中引入，Laravel 13 中仍然可用。默认驱动使用子 PHP 进程，因此无需额外的扩展包即可运行。

工作原理

Concurrency Facade 会将传入的闭包序列化后发送给一个隐藏的 Artisan 命令，并让它们各自作为独立的 PHP 进程执行。处理完成后，会将返回值序列化后返回给父进程。它提供了三种驱动。

驱动	说明
`process`	默认驱动。启动子 PHP 进程执行，可在 Web 请求中运行
`fork`	需要 `spatie/fork` 扩展包。通过 fork 进程运行，仅可在 CLI 中使用，但速度更快
`sync`	不并行执行，顺序运行。适合测试场景

基本用法

Concurrency::run()

向 run() 方法传入闭包数组，这些闭包会被并行执行。返回值是各闭包返回值组成的数组。

use Illuminate\Support\Facades\Concurrency;
use Illuminate\Support\Facades\DB;

[$userCount, $orderCount] = Concurrency::run([
    fn () => DB::table('users')->count(),
    fn () => DB::table('orders')->count(),
]);

可以通过数组解构把每个结果分别赋值给变量。闭包的顺序与返回值顺序一致。

指定驱动

如果希望使用特定的驱动，可以通过 driver() 方法指定。

$results = Concurrency::driver('fork')->run([
    fn () => fetchFromApiA(),
    fn () => fetchFromApiB(),
]);

要修改默认驱动，可以先发布配置文件，然后更新 default 选项。

php artisan config:publish concurrency

fork 驱动的用法

fork 驱动比 process 驱动更快，但只能在 PHP 的 CLI 环境（Artisan 命令或队列 worker）中使用。 Web 请求中无法使用。使用前请先安装 spatie/fork 扩展包。

composer require spatie/fork

fork 驱动在 Web 请求中无法工作。请在 Artisan 命令或队列 worker 中进行并发处理时使用它。

不接收执行结果：Concurrency::defer()

如果你并不关心处理结果，希望在返回 HTTP 响应之后再在后台执行，可以使用 defer() 方法。

use App\Services\Metrics;
use Illuminate\Support\Facades\Concurrency;

Concurrency::defer([
    fn () => Metrics::report('users'),
    fn () => Metrics::report('orders'),
]);

调用 defer() 时闭包不会立即执行。它们会在 HTTP 响应发送给用户之后并行执行。

埋点数据记录、缓存预热等不需要让用户等待的处理，非常适合使用 defer()。

实战示例：同时调用多个外部 API

以电商网站的仪表盘为例，需要从库存管理 API、销售统计 API、物流状态 API 三个接口获取信息。

顺序执行（改造前）

use Illuminate\Support\Facades\Http;

public function dashboard(): array
{
    // 每次请求都要等待前一个完成（合计大约 3 秒）
    $inventory = Http::get('https://api.example.com/inventory')->json();
    $sales     = Http::get('https://api.example.com/sales')->json();
    $shipping  = Http::get('https://api.example.com/shipping')->json();

    return compact('inventory', 'sales', 'shipping');
}

并行执行（改造后）

use Illuminate\Support\Facades\Concurrency;
use Illuminate\Support\Facades\Http;

public function dashboard(): array
{
    // 3 个请求同时执行（总耗时约等于最慢的那个 API）
    [$inventory, $sales, $shipping] = Concurrency::run([
        fn () => Http::get('https://api.example.com/inventory')->json(),
        fn () => Http::get('https://api.example.com/sales')->json(),
        fn () => Http::get('https://api.example.com/shipping')->json(),
    ]);

    return compact('inventory', 'sales', 'shipping');
}

如果每个 API 各需要 1 秒，顺序执行总共要 3 秒，而并行执行大约 1 秒即可完成。

同时执行多条数据库聚合

use Illuminate\Support\Facades\Concurrency;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Cache;

public function statistics(): array
{
    [$totalUsers, $activeUsers, $totalOrders, $revenue] = Concurrency::run([
        fn () => DB::table('users')->count(),
        fn () => DB::table('users')->where('active', true)->count(),
        fn () => DB::table('orders')->count(),
        fn () => DB::table('orders')->sum('total_amount'),
    ]);

    return [
        'total_users'  => $totalUsers,
        'active_users' => $activeUsers,
        'total_orders' => $totalOrders,
        'revenue'      => $revenue,
    ];
}

用 process 驱动执行数据库聚合时，每个子进程都会新建一个数据库连接。并发数较高时请注意数据库的最大连接数。

Context 的自动传播

在 Laravel 13.31（#61419）中，process 驱动会自动把当前的 Context 传播到子进程。在此之前，通过 Concurrency::run() 或 Concurrency::defer() 启动的子进程无法继承 Context::add() 设置的值，日志中输出的 context 会在父进程与子进程之间出现差异。修复后，内部使用与队列任务传递上下文相同的机制。

启动子进程之前，将 Context::dehydrate() 的结果序列化为 JSON，作为 __LARAVEL_CONTEXT 环境变量传入
子进程（隐藏的 Artisan 命令 invoke-serialized-closure）启动后读取 __LARAVEL_CONTEXT，并对 Context 仓库执行 hydrate()

因此，可以在闭包内直接引用在闭包外部设置的 Context。

use Illuminate\Support\Facades\Concurrency;
use Illuminate\Support\Facades\Context;
use Illuminate\Support\Facades\Http;
use Illuminate\Support\Facades\Log;
use Illuminate\Support\Str;

Context::add('request_id', (string) Str::uuid());

Concurrency::run([
    function () {
        // 可以直接取得父进程中设置的 request_id
        Log::info('inventory fetched', ['request_id' => Context::get('request_id')]);

        return Http::get('https://api.example.com/inventory')->json();
    },
]);

fork 驱动使用 spatie/fork 提供的操作系统级 fork()，子进程会直接复制父进程的内存空间来启动。因此 Context 本来就自然共享，这次的修复主要针对 process 驱动（另起进程的方式）。

只要把请求 ID、用户 ID 等用于日志关联的值放入 Context，就能以相同的键跟踪 Concurrency::run() 并发执行的处理产生的日志。详细用法请参考共享上下文。

测试环境的配置

在测试环境中可以使用 sync 驱动，让闭包顺序执行。省去了并行进程的启动开销，测试可以更快完成。

// tests/Feature/DashboardTest.php
use Illuminate\Support\Facades\Concurrency;

public function test_dashboard_returns_correct_data(): void
{
    Concurrency::fake(); // 切换到 sync 驱动

    $response = $this->get('/dashboard');

    $response->assertStatus(200);
}

或者在 .env.testing 中修改默认驱动。

CONCURRENCY_DRIVER=sync

注意事项

闭包的限制

闭包会被序列化传给子进程。不能从闭包外部捕获无法序列化的对象（数据库连接、文件句柄、资源等）。需要的对象请在闭包内部重新创建。

// NG：从外部捕获 Eloquent 模型
$user = User::find(1);
Concurrency::run([
    fn () => $user->orders()->count(), // 可能无法序列化
]);

// OK：在闭包内部获取数据
$userId = 1;
Concurrency::run([
    fn () => Order::where('user_id', $userId)->count(),
]);

process 驱动的开销

process 驱动需要启动子进程，因此如果处理本身非常短（几毫秒以内），并行执行未必更快。 HTTP 请求、数据库聚合等耗时 100ms 以上的处理，通过并行化才能真正体现出效果。

fork 驱动的限制

fork 驱动会 fork PHP 进程，因此无法在 Web 请求（FPM 或 Apache）中使用。请只在 Artisan 命令或队列 worker 中使用它。

异常的处理

并行执行中若发生异常，run() 会将异常重新抛出。如果希望某个任务出错时其他任务仍能继续执行，请在闭包内使用 try/catch。

Concurrency::run([
    function () {
        try {
            return Http::get('https://api.example.com/data')->json();
        } catch (\Exception $e) {
            return null; // 失败时返回 null
        }
    },
]);

下一步

队列与任务

了解如何使用队列与任务在后台异步执行处理

架构

数据库

Eloquent

安全

事件

邮件与通知

支付

异步处理

实时通信

调度

缓存与优化

国际化

文件与存储

调试与监控

控制台与 CLI

性能

前端

AI 开发

开发环境

什么是并发处理

工作原理

基本用法

Concurrency::run()

指定驱动

fork 驱动的用法

不接收执行结果：Concurrency::defer()

实战示例：同时调用多个外部 API

顺序执行（改造前）

并行执行（改造后）

同时执行多条数据库聚合

Context 的自动传播

测试环境的配置

注意事项

下一步

队列与任务

​什么是并发处理

​工作原理

​基本用法

​Concurrency::run()

​指定驱动

​fork 驱动的用法

​不接收执行结果：Concurrency::defer()

​实战示例：同时调用多个外部 API

​顺序执行（改造前）

​并行执行（改造后）

​同时执行多条数据库聚合

​Context 的自动传播

​测试环境的配置

​注意事项

​下一步

队列与任务

什么是并发处理

工作原理

基本用法

Concurrency::run()

指定驱动

fork 驱动的用法

不接收执行结果：Concurrency::defer()

实战示例：同时调用多个外部 API

顺序执行（改造前）

并行执行（改造后）

同时执行多条数据库聚合

Context 的自动传播

测试环境的配置

注意事项

下一步