PHP 管道/流水线/Pipeline模式

• 作用

  • 其实Pipeline模式和装饰者模式类似
  • 管道就是把一系列串联执行的程序按顺序分工处理

• 举例

  • 顾客在商城提交商品创建订单、支付很简单常见;
  • 现在商城新加了一个会员卡，一级会员打 9.5 折，二级会员 9 折，三级会员 8.5 折;
  • 随后需求又要加入优惠券系统，部分商品需要打折；
  • 如果把这些串联执行的程序放在一起处理，代码程序看起来会越来越臃肿不堪，难以维护；
  • 老员工离职，新员工入职接手这样的代码可能直接会疯掉；

• 解决

  • 如果把一系列串联执行的程序按顺序分工处理，按串联顺序把一系列的创建订单、会员卡打折、优惠券等等功能分别处理，这样就会很nice。
  • 简单来说就是 将「输入」引入管道，根据每个小任务对输入进行操作 (加工、过滤)，最后输出满足需要的结果。
• 有对装饰者模式感兴趣的同学可以看这篇文章

• 有对设计模式感兴趣的可以看这篇文章

代码演示1

<?php

interface PipelineInterface
{
    public function __construct($payLoad);

    public function pipe(StageInterface $stage);

    public function process();
}

interface StageInterface
{
    public function handle($payLoad);
}

class StageAddHundred implements StageInterface
{
    public function handle($payLoad)
    {
        return $payLoad + 100;
    }
}

class StageMultiHundred implements StageInterface
{
    public function handle($payLoad)
    {
        return $payLoad * 100;
    }
}

class Pipeline implements PipelineInterface
{
    private $pipes;

    private $payLoad;

    public function __construct($payLoad)
    {
        $this->payLoad = $payLoad;
    }

    public function pipe(StageInterface $stage)
    {
        $this->pipes[] = $stage;
        return $this;
    }

    public function process()
    {
        foreach ($this->pipes as $eachPipe) {
            $this->payLoad = call_user_func([$eachPipe, 'handle'], $this->payLoad);
        }

        return $this->payLoad;
    }
}

//int(11000)
$pipe = (new Pipeline(10))
    ->pipe(new StageAddHundred())
    ->pipe(new StageMultiHundred())
    ->process();

代码演示2

• 推荐一个 PHP 插件：league/pipeline

<?php
   
   include 'vendor/autoload.php';
   
   use League\Pipeline\Pipeline;
   use League\Pipeline\StageInterface;
   
   class OrderStage implements StageInterface
   {
       public function __invoke($payload)
       {
           $payload['order'] = ['info' => '我是订单详情'];
           return   $payload;
       }
   }
   
   class Stage implements StageInterface
   {
       public function __invoke($payload)
       {
          $payload['coupon'] = ['info'=>'我是优惠信息'];
           return $payload;
       }
   }
   

    $pipeline = (new Pipeline)
    ->pipe(new OrderStage)
    ->pipe(new CouponStage);
   
   $pipeline->process(array());

   /*Array
   (
       [order] => Array
       (
           [info] => 我是订单详情
       )
   
       [coupon] => Array
   (
       [info] => 我是优惠信息
   )
        
        )*/

拓展 redis中的pipeline

• 众所周知,redis是单线程的，循环执行n个命令和打包执行n个命令对比如下

图解

代码直观对比

• php操作redis(pipeline)代码

<?php


$stime = microtime(true); //获取程序开始执行的时间
echo '开始内存：'.$sm = memory_get_usage(), '<hr>';
echo PHP_EOL;
$redis = new \Redis();

$redis->connect('127.0.0.1', 6379);


$pipe = $redis->multi($redis::PIPELINE);

for ($i = 0; $i < 10000; $i++) {

    $pipe->set("key::$i", str_pad($i, 4, '0', 0));

    $pipe->get("key::$i");

}

$replies = $pipe->exec();

$etime = microtime(true);//获取程序执行结束的时间

$total = ($etime - $stime);   //计算差值


echo "[页面执行时间：{$total}s<hr>";
echo '运行后内存：'.$em = memory_get_usage(), '<hr>';
//开始内存：393576
//[页面执行时间：0.070788860321045s
//运行后内存：1775136

• php操作redis代码

<?php
  $stime=microtime(true); //获取程序开始执行的时间
  echo '开始内存：'.$sm = memory_get_usage(), '<hr>';
  $redis = new \Redis();
  
  $redis->connect('127.0.0.1',6379);
  
  
  
  $t1 = time();
  for($i= 0; $i<10000 ; $i++) {
  
      $redis->set("key::$i",str_pad($i,4,'0',0));
  
      $redis->get("key::$i");
  
  }
  
  
  $etime=microtime(true);//获取程序执行结束的时间
  
  $total=($etime-$stime);   //计算差值
  
  
  echo "[页面执行时间：{$total}s<hr>";
  echo '运行后内存：'.$em = memory_get_usage(), '<hr>';

//开始内存：392968
//[页面执行时间：2.7398328781128s
//运行后内存：401800