Pipeline 模式

什么是 Pipeline 模式

Pipeline 模式是一种将被处理对象（passable）依次通过一系列管道（处理步骤）的设计模式。每个管道接收上一步的输出、施加处理后再传给下一步。在 Laravel 中，Illuminate\Pipeline\Pipeline 类实现了该模式，中间件的处理正是通过这种机制工作的。

请求 → [中间件 1] → [中间件 2] → [Controller] → 响应

在 Laravel 核心中的使用位置

Pipeline 在 Laravel 内部被广泛使用。

HTTP 中间件 — Illuminate\Foundation\Http\Kernel 将请求送入中间件 Pipeline
控制台命令 — Artisan 命令的前后处理
Router 中间件 — 应用于路由组的中间件

来看看 HTTP Kernel 使用 Pipeline 的部分：

// 摘自 Illuminate\Foundation\Http\Kernel
protected function sendRequestThroughRouter($request)
{
    return (new Pipeline($this->app))
        ->send($request)
        ->through($this->app->shouldSkipMiddleware() ? [] : $this->middleware)
        ->then($this->dispatchToRouter());
}

基本用法

send / through / thenReturn

最简单的 Pipeline 结构。

use Illuminate\Pipeline\Pipeline;

$result = app(Pipeline::class)
    ->send('hello')
    ->through([
        function (string $passable, Closure $next): string {
            return $next(strtoupper($passable));
        },
        function (string $passable, Closure $next): string {
            return $next($passable . '!');
        },
    ])
    ->thenReturn();

// $result === 'HELLO!'

send($passable) — 指定要送入 Pipeline 的对象
through($pipes) — 指定管道数组
thenReturn() — 执行 Pipeline 并返回结果

then

若要用闭包指定最终处理，使用 then()。

$result = app(Pipeline::class)
    ->send($request)
    ->through([AuthPipe::class, LogPipe::class])
    ->then(function ($request) {
        return 'processed: ' . $request;
    });

pipe

要在之后追加管道，使用 pipe()。

$pipeline = app(Pipeline::class)
    ->send($data)
    ->through([FirstPipe::class]);

// 按条件插入额外管道
if ($needsExtra) {
    $pipeline->pipe(ExtraPipe::class);
}

$result = $pipeline->thenReturn();

基于类的管道

用专用类替代闭包，可以提高可复用性。每个管道类都实现 handle 方法。

namespace App\Pipes;

use Closure;

class TrimPipe
{
    public function handle(string $passable, Closure $next): string
    {
        return $next(trim($passable));
    }
}

namespace App\Pipes;

use Closure;

class SanitizePipe
{
    public function handle(string $passable, Closure $next): string
    {
        $sanitized = htmlspecialchars($passable, ENT_QUOTES, 'UTF-8');

        return $next($sanitized);
    }
}

use App\Pipes\TrimPipe;
use App\Pipes\SanitizePipe;
use Illuminate\Pipeline\Pipeline;

$result = app(Pipeline::class)
    ->send('  <script>alert("xss")</script>  ')
    ->through([TrimPipe::class, SanitizePipe::class])
    ->thenReturn();

// $result === '&lt;script&gt;alert(&quot;xss&quot;)&lt;/script&gt;'

via — 改变调用的方法

默认调用管道的 handle 方法，可通过 via() 指定别的方法名。

class ValidatePipe
{
    public function process(array $data, Closure $next): array
    {
        // 验证逻辑
        return $next($data);
    }
}

$result = app(Pipeline::class)
    ->send($data)
    ->through([ValidatePipe::class])
    ->via('process')
    ->thenReturn();

向管道传参

可以通过在类名后加 : 与 , 向管道传参。这与中间件的 throttle:60,1 语法机制相同。

namespace App\Pipes;

use Closure;

class LimitLengthPipe
{
    public function handle(string $passable, Closure $next, int $max = 100): string
    {
        return $next(substr($passable, 0, $max));
    }
}

$result = app(Pipeline::class)
    ->send($longText)
    ->through(['App\Pipes\LimitLengthPipe:50'])
    ->thenReturn();

finally — Pipeline 结束后的处理

使用 finally() 方法可以注册在 Pipeline 结束后一定会执行的回调。无论成功还是失败都会执行。

$result = app(Pipeline::class)
    ->send($request)
    ->through([LogPipe::class, AuthPipe::class])
    ->finally(function ($passable) {
        // 记录日志、释放资源等
        logger()->info('Pipeline completed', ['request' => $passable]);
    })
    ->thenReturn();

withinTransaction — 在事务中执行

Laravel 11 以后，可以让整个 Pipeline 在数据库事务中执行。

$result = app(Pipeline::class)
    ->send($order)
    ->through([
        CreateOrderPipe::class,
        ChargePaymentPipe::class,
        SendConfirmationPipe::class,
    ])
    ->withinTransaction()
    ->thenReturn();

也可以指定特定的数据库连接。

->withinTransaction('mysql')

任何管道抛出异常时，整个事务都会回滚。

实战示例：订单处理 Pipeline

将电商站点的订单处理组合成多个管道的例子。

创建管道类

namespace App\Pipes\Order;

use App\Models\Order;
use Closure;

class ValidateInventoryPipe
{
    public function handle(Order $order, Closure $next): Order
    {
        foreach ($order->items as $item) {
            if ($item->product->stock < $item->quantity) {
                throw new \RuntimeException("库存不足：{$item->product->name}");
            }
        }

        return $next($order);
    }
}

namespace App\Pipes\Order;

use App\Models\Order;
use Closure;

class ReserveInventoryPipe
{
    public function handle(Order $order, Closure $next): Order
    {
        foreach ($order->items as $item) {
            $item->product->decrement('stock', $item->quantity);
        }

        return $next($order);
    }
}

namespace App\Pipes\Order;

use App\Models\Order;
use App\Services\PaymentService;
use Closure;

class ProcessPaymentPipe
{
    public function __construct(
        protected PaymentService $payment,
    ) {}

    public function handle(Order $order, Closure $next): Order
    {
        $this->payment->charge($order->total, $order->payment_token);
        $order->update(['status' => 'paid']);

        return $next($order);
    }
}

执行 Pipeline

use App\Models\Order;
use App\Pipes\Order\ValidateInventoryPipe;
use App\Pipes\Order\ReserveInventoryPipe;
use App\Pipes\Order\ProcessPaymentPipe;
use Illuminate\Pipeline\Pipeline;

class OrderService
{
    public function process(Order $order): Order
    {
        return app(Pipeline::class)
            ->send($order)
            ->through([
                ValidateInventoryPipe::class,
                ReserveInventoryPipe::class,
                ProcessPaymentPipe::class,
            ])
            ->withinTransaction()
            ->thenReturn();
    }
}

Pipeline 的内部实现

carry() 方法是 Pipeline 的核心。它借助 array_reduce 从右往左折叠各个管道，构建闭包链。

// Pipeline::then() 的内部
protected function carry()
{
    return function ($stack, $pipe) {
        return function ($passable) use ($stack, $pipe) {
            if (is_callable($pipe)) {
                return $pipe($passable, $stack);
            } elseif (! is_object($pipe)) {
                [$name, $parameters] = $this->parsePipeString($pipe);
                $pipe = $this->getContainer()->make($name);
                $parameters = array_merge([$passable, $stack], $parameters);
            } else {
                $parameters = [$passable, $stack];
            }

            return $pipe->{$this->method}(...$parameters);
        };
    };
}

先用 array_reverse 反转管道，再交给 array_reduce，以保证调用顺序正确。为了让第一个管道最先执行，从内向外构建闭包栈。

Pipeline 也被称作「洋葱型」架构。请求从外层（第一个管道）向内部通过，响应则从内部返回到外部。因此每个中间件都可以在 $next() 之前和之后加上处理。

与 Macroable trait 组合使用

Pipeline 类使用了 Macroable trait，可以添加自定义方法。

use Illuminate\Pipeline\Pipeline;

Pipeline::macro('sendThroughLogging', function (array $pipes) {
    /** @var Pipeline $this */
    return $this->through(array_map(function ($pipe) {
        return function ($passable, $next) use ($pipe) {
            logger()->debug("Entering pipe: {$pipe}");
            $result = $next($passable);
            logger()->debug("Exiting pipe: {$pipe}");
            return $result;
        };
    }, $pipes));
});

$result = app(Pipeline::class)
    ->send($data)
    ->sendThroughLogging([TrimPipe::class, SanitizePipe::class])
    ->thenReturn();

下一步

Macroable trait

学习向既有类添加自定义方法的 Macroable trait 的用法。

AI 开发

设计模式

包开发

应用结构

Laravel 13 新特性

Eloquent 深入

验证扩展

认证扩展

安全

测试

什么是 Pipeline 模式

在 Laravel 核心中的使用位置

基本用法

send / through / thenReturn

then

pipe

基于类的管道

via — 改变调用的方法

向管道传参

finally — Pipeline 结束后的处理

withinTransaction — 在事务中执行

实战示例：订单处理 Pipeline

Pipeline 的内部实现

与 Macroable trait 组合使用

下一步

Macroable trait

​什么是 Pipeline 模式

​在 Laravel 核心中的使用位置

​基本用法

​send / through / thenReturn

​then

​pipe

​基于类的管道

​via — 改变调用的方法

​向管道传参

​finally — Pipeline 结束后的处理

​withinTransaction — 在事务中执行

​实战示例：订单处理 Pipeline

​Pipeline 的内部实现

​与 Macroable trait 组合使用

​下一步

Macroable trait

什么是 Pipeline 模式

在 Laravel 核心中的使用位置

基本用法

send / through / thenReturn

then

pipe

基于类的管道

via — 改变调用的方法

向管道传参

finally — Pipeline 结束后的处理

withinTransaction — 在事务中执行

实战示例：订单处理 Pipeline

Pipeline 的内部实现

与 Macroable trait 组合使用

下一步