当前位置：首页 > PHP

php管道实现

2026-01-29 18:06:26PHP

PHP 管道模式实现

管道模式（Pipeline Pattern）是一种将任务分解为多个步骤的设计模式，每个步骤由独立的处理器完成，数据通过管道依次传递。PHP 中可以通过多种方式实现管道模式。

基础实现方式

使用闭包和数组实现简单的管道处理：

$pipeline = array_reduce(
    [
        function ($input) { return $input * 2; },
        function ($input) { return $input + 10; },
        function ($input) { return $input / 3; }
    ],
    function ($carry, $callback) {
        return $callback($carry);
    },
    5 // 初始输入值
);

echo $pipeline; // 输出: 6.6666666666667

面向对象实现

创建可扩展的管道类：

class Pipeline
{
    private $stages = [];

    public function add(callable $stage): self
    {
        $this->stages[] = $stage;
        return $this;
    }

    public function process($payload)
    {
        foreach ($this->stages as $stage) {
            $payload = $stage($payload);
        }
        return $payload;
    }
}

// 使用示例
$pipeline = (new Pipeline())
    ->add(function ($x) { return $x * 2; })
    ->add(function ($x) { return $x + 10; })
    ->add(function ($x) { return $x / 3; });

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

Laravel 风格的管道实现

Laravel 框架提供了更完善的管道实现：

interface Pipe
{
    public function handle($passable, \Closure $next);
}

class Double implements Pipe
{
    public function handle($passable, \Closure $next)
    {
        return $next($passable * 2);
    }
}

class AddTen implements Pipe
{
    public function handle($passable, \Closure $next)
    {
        return $next($passable + 10);
    }
}

class DivideByThree implements Pipe
{
    public function handle($passable, \Closure $next)
    {
        return $next($passable / 3);
    }
}

function pipeline($passable, array $pipes)
{
    $carry = array_reduce(
        array_reverse($pipes),
        function ($stack, $pipe) {
            return function ($passable) use ($stack, $pipe) {
                return $pipe->handle($passable, $stack);
            };
        },
        function ($passable) {
            return $passable;
        }
    );

    return $carry($passable);
}

// 使用示例
$result = pipeline(5, [
    new Double(),
    new AddTen(),
    new DivideByThree()
]);

echo $result; // 输出: 6.6666666666667

中间件式管道

适合处理请求/响应场景的管道：

class Pipeline
{
    private $pipes = [];

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

    public function then(\Closure $destination)
    {
        $pipeline = array_reduce(
            array_reverse($this->pipes),
            $this->carry(),
            $destination
        );

        return $pipeline();
    }

    protected function carry(): \Closure
    {
        return function ($stack, $pipe) {
            return function () use ($stack, $pipe) {
                return $pipe->handle(request(), $stack);
            };
        };
    }
}

// 使用示例
$response = (new Pipeline)
    ->through([
        new Authenticate(),
        new CheckForMaintenanceMode(),
        new AddQueuedCookiesToResponse(),
    ])
    ->then(function ($request) {
        return dispatchToRouter($request);
    });

异步管道实现

使用生成器实现异步管道处理：

php管道实现

function asyncPipeline($initial, array $stages)
{
    $current = $initial;

    foreach ($stages as $stage) {
        $current = yield $stage($current);
    }

    return $current;
}

// 使用示例
$coroutine = asyncPipeline(5, [
    function ($x) { return $x * 2; },
    function ($x) { return $x + 10; },
    function ($x) { return $x / 3; }
]);

$result = $coroutine->current(); // 10
$result = $coroutine->send($result); // 20
$result = $coroutine->send($result); // 6.666...