php实现pipeline

PHP 实现 Pipeline 模式

Pipeline（管道）模式是一种将多个处理步骤串联起来的设计模式，每个步骤处理输入并传递给下一个步骤。在 PHP 中可以通过多种方式实现。

基础实现方式

使用闭包和数组包装步骤，通过循环依次调用：

$pipeline = [
    function ($input) { return $input * 2; },
    function ($input) { return $input + 10; },
    function ($input) { return $input / 3; }
];

$result = 5;
foreach ($pipeline as $step) {
    $result = $step($result);
}
echo $result; // 输出: 6.666...

面向对象实现

创建 Pipeline 类管理步骤和执行流程：

class Pipeline
{
    private $steps = [];

    public function addStep(callable $step)
    {
        $this->steps[] = $step;
        return $this;
    }

    public function process($input)
    {
        foreach ($this->steps as $step) {
            $input = $step($input);
        }
        return $input;
    }
}

// 使用示例
$pipeline = new Pipeline();
$pipeline->addStep(fn($x) => $x * 2)
         ->addStep(fn($x) => $x + 10)
         ->addStep(fn($x) => $x / 3);

echo $pipeline->process(5); // 输出: 6.666...

中间件风格的实现

适用于请求处理场景，每个步骤可以决定是否中断管道：

class MiddlewarePipeline
{
    private $middlewares = [];
    private $index = 0;

    public function addMiddleware(callable $middleware)
    {
        $this->middlewares[] = $middleware;
    }

    public function handle($request)
    {
        if (!isset($this->middlewares[$this->index])) {
            return $request;
        }

        $middleware = $this->middlewares[$this->index];
        $this->index++;

        return $middleware($request, fn($req) => $this->handle($req));
    }
}

// 使用示例
$pipeline = new MiddlewarePipeline();
$pipeline->addMiddleware(function ($req, $next) {
    $req['user'] = 'admin';
    return $next($req);
});
$pipeline->addMiddleware(function ($req, $next) {
    $req['logged'] = true;
    return $next($req);
});

$result = $pipeline->handle([]);
print_r($result); // 输出包含 user 和 logged 的数组

Laravel 风格的 Pipeline

模仿 Laravel 框架的实现方式：

class LaravelPipeline
{
    protected $passable;
    protected $pipes = [];

    public function send($passable)
    {
        $this->passable = $passable;
        return $this;
    }

    public function through(array $pipes)
    {
        $this->pipes = $pipes;
        return $this;
    }

    public function then(callable $destination)
    {
        $pipeline = array_reduce(
            array_reverse($this->pipes),
            $this->carry(),
            fn($passable) => $destination($passable)
        );

        return $pipeline($this->passable);
    }

    protected function carry()
    {
        return function ($stack, $pipe) {
            return function ($passable) use ($stack, $pipe) {
                return $pipe($passable, $stack);
            };
        };
    }
}

// 使用示例
$pipeline = new LaravelPipeline();
$result = $pipeline->send(5)
    ->through([
        function ($passable, $next) { return $next($passable * 2); },
        function ($passable, $next) { return $next($passable + 10); },
        function ($passable, $next) { return $next($passable / 3); }
    ])
    ->then(function ($result) {
        return $result;
    });

echo $result; // 输出: 6.666...

异常处理增强版

为管道添加异常处理能力：

class SafePipeline
{
    private $steps = [];
    private $handlers = [];

    public function addStep(callable $step)
    {
        $this->steps[] = $step;
        return $this;
    }

    public function addHandler(string $exceptionClass, callable $handler)
    {
        $this->handlers[$exceptionClass] = $handler;
        return $this;
    }

    public function process($input)
    {
        try {
            foreach ($this->steps as $step) {
                $input = $step($input);
            }
            return $input;
        } catch (Throwable $e) {
            foreach ($this->handlers as $exceptionClass => $handler) {
                if ($e instanceof $exceptionClass) {
                    return $handler($e, $input);
                }
            }
            throw $e;
        }
    }
}

// 使用示例
$pipeline = new SafePipeline();
$pipeline->addStep(fn($x) => $x * 2)
         ->addStep(function ($x) {
             if ($x > 100) throw new RuntimeException('Value too large');
             return $x + 10;
         })
         ->addHandler(RuntimeException::class, fn($e, $x) => $x / 2);

echo $pipeline->process(50); // 输出: 55 (50*2=100, +10=110 → 触发异常 → 110/2=55)

这些实现方式展示了 PHP 中 Pipeline 模式的不同应用场景和变体，可以根据具体需求选择合适的实现方案。