PHP异步编程探索

39 阅读 0 评论 0点赞

image.png

我们知道，PHP 最初是为了支持同步开发而创建的，因此大多数 PHP 开发人员习惯于仅使用php编写同步代码，那么我们能用Php来进行异步编程吗。答案是可以的。

什么是异步编程

当我们用php编写的应用程序I/O任务时，程序会在执行某个任务之前，一定要等待之前的任务完成，这时CPU会有很多时间处于空闲状态，这不仅会降低应用程序性能，还会降低硬件利用率。比如，当程序需要从数据库中读取大量的数据时，由于需要等待I/O操作完成，程序的执行速度会非常缓慢。因此，我们通过Swoole扩展或者一些如Reactphp的库，在程序执行的过程中，不需要等待某个任务完成才能执行下一个任务。这种编程模式可以极大地提高程序的效率和响应速度，尤其在处理复杂的I/O操作时表现得更为出色，而这就是异步编程。

Php pcntl扩展

通过异步编程的概念可知，异步编程的可以通过多进程来实现，在php中，pcntl扩展是首选的用来处理多进程的库，我们先测一段同步的代码

<?php
class AsyncPhp{
    public function curl($url)
    {
        $ch = curl_init();
        curl_setopt($ch, CURLOPT_URL, $url);
        curl_setopt($ch, CURLOPT_RETURNTRANSFER, 1);
        curl_setopt($ch, CURLOPT_HEADER, 0);
        $output = curl_exec($ch);
        curl_close($ch);
        return $output;
    }
    public function testSync(){
        $url = 'https://www.baidu.com/?i=';
        $start=microtime(true);    
        for($i=0;$i<5;$i++){
            $this->curl($url.$i);
            echo '访问第'.($i+1).'次百度'."\n";
        }
        $end=microtime(true);

        echo "\n总用时",$end-$start."\n";        
    }
}

$tester = new AsyncPhp();
$tester->testSync();

当我们同步请求5次百度时，是0.9秒

这时为了提高性能，pcntl就可以派上用场。
pcntl是一个可以利用操作系统的fork系统调用在PHP中实现多线程的进程控制扩展，当使用fork系统调用后执行的代码将会是并行的。pcntl的缺点是只适用于Linux平台的CLI模式下使用
我们在刚才的测试来新增如下方法来测试

    public function testAsync(){
        $url = 'https://www.baidu.com/?i=';
        $start=microtime(true); 
        $i=0;
        while($i<5){
            $pids[$i] = pcntl_fork(); //创建子进程
             if($pids[$i] == 0){ //返回0表示在子进程
                 $this->curl($url.$i);; //子进程执行代码
                 exit(0);
             }
             $i++;
         }
         //等待进程关闭
         for($i=0;$i<5;$i++){
             pcntl_waitpid($pids[$i],$status,WUNTRACED);//等待进程结束
             if(pcntl_wifexited($status)){
                 //子进程完成退出
                 echo '第'.$i.'次访问百度，用时:'.microtime(true)-$start."\n";
             }
         } 

        $end=microtime(true);

        echo "\n总用时",$end-$start."\n";        
    }

在上面的代码中，当程序运行pcntl_fork时，Linux系统立fork出一个子进程，并在子进程中返回0，所以$this->curl是在子进程中运行的，exit(0）是终止了这个子进程，为什么要终止？如果不终止，子进程就会继续执行while这样最终导致产生很多次访问，而我们只要5次。pcntl_waitpid用于等待子进程结束，这样就可以计算子进程运行时间了。代码执行结果如下

通过对比可知，php使用pcntl扩展实现异步总用时比传统同步方式少用时0.7秒，性能提升了3倍多。但是pcntl只能跑在cli环境下，传统php-fpm环境下是无法使用的，这时我们可以在网络请求这边想办法，于是有了curl_multi。

使用curl_multi函数组

**curl_multi函数组允许异步处理多个 cURL 句柄,**下面是具体代码

    public function getCurl($url)
    {
        $ch = curl_init();
        curl_setopt($ch, CURLOPT_URL, $url);
        curl_setopt($ch, CURLOPT_RETURNTRANSFER, 1);
        curl_setopt($ch, CURLOPT_HEADER, 0);
        return $ch;
    }    
    public function testCurlMulti(){
        $url = 'https://www.baidu.com/?i=';
        $start=microtime(true); 
        $mh = curl_multi_init();

        $curl_array = array();
        for ($i=0; $i < 5; $i++) { 
            $curl_array[$i] = $this->getCurl($url.$i);
            curl_multi_add_handle($mh, $curl_array[$i]);
        }
        $running = NULL;
        do {
            curl_multi_exec($mh,$running);
        } while($running > 0);

        foreach($curl_array as $i=>$item){
            $cotnent = curl_multi_getcontent($curl_array[$i]);
            echo '第'.$i.'次访问百度，用时:'.microtime(true)-$start."\n";
            curl_multi_remove_handle($mh, $curl_array[$i]);
        }
        curl_multi_close($mh);  

        $end=microtime(true);

        echo "\n总用时",$end-$start."\n";            
    }

在上面的代码中，curl_multi_exec函数将并行执行每个curl句柄。最终总用时如下：

也比同步的代码快了一倍了。
curl_multi函数组在很多场景下都有使用到，特别是在网络爬虫开发中会经常使用到。

Parallel扩展

同pcntl一样，Parallel^[1]也是一个php扩展，安装方法：https://github.com/krakjoe/parallel/blob/develop/INSTALL.md，如果安装时报如下错误，

则需要重新编译php，编译方式如下：

wget https://www.php.net/distributions/php-8.1.1.tar.gz
tar -zxvf ./php-8.1.1.tar.gz
cd php-8.1.1
# 编译
./configure --prefix=/usr/local/php81 --enable-debug --enable-zts --with-config-file-path=/usr/local/php81/etc --with-config-file-scan-dir=/usr/local/php81/conf.d --with-mysqli=mysqlnd --with-pdo-mysql=mysqlnd --with-iconv=/usr/local --with-freetype=/usr/local/freetype --with-jpeg --with-zlib --enable-xml --disable-rpath --enable-bcmath --enable-shmop --enable-sysvsem --with-curl --enable-mbregex --enable-mbstring --enable-intl --enable-pcntl --enable-ftp --enable-gd --with-openssl --with-mhash --enable-pcntl --enable-sockets --with-zip --enable-soap --with-gettext --enable-opcache --with-xsl --with-pear --disable-mbregex
make && make install
# 搞定
ln -s /usr/local/php81/bin/php /usr/bin/php81
php -v

以上安装过程中有可能会出错，出错的解决方法请看我的另一篇文章《为了编译启用ZTS的php，我一路披荆斩棘》。
成功安装php之后,会打印如下结果

ZTS的php环境有了，就可以安装parallel了

git clone https://github.com/krakjoe/parallel.git
cd parallel
phpize
./configure --enable-parallel  [ --enable-parallel-coverage ] [ --enable-parallel-dev ]
make
make test
make install

安装成功后是这样子的

我们继续新增testParallel方法。代码如下

    public function testParallel(){        
        $ch = new Channel();
        
        // executed within thread
        $start=microtime(true); 
        $task = function (Channel $channel, int $i){
            $url = 'https://www.baidu.com/?i=';
            $url = $url.$i;
            $ch = curl_init();
            curl_setopt($ch, CURLOPT_URL, $url);
            curl_setopt($ch, CURLOPT_RETURNTRANSFER, 1);
            curl_setopt($ch, CURLOPT_HEADER, 0);
            $output = curl_exec($ch);
            curl_close($ch);
            // echo $output;
            echo "第".$i."次访问百度\n";
              $channel->send($output);       
        };
        
        // creating a few threads
        $runtimeList = [];
        for ($i = 0; $i < 5; $i++) {
            $runtimeList[] = new Runtime();
        }
        // run all threads
        $futureList = [];
        foreach ($runtimeList as $i => $runtime) {
            $futureList[] = $runtime->run($task, [$ch, $i]);
        }
        
        $ch->close();
        
        // echo $queue . PHP_EOL;
    }

Parallel底层用的也是多进程机制，但术语一般是说成线程，上面的代码中，我们定义了5个线程，并发运行结果如下

ReactPHP

与前面几个扩展库不同，ReactPHP^[2] 是一个用PHP编写的基于事件驱动编程（EventLoop)的低级库。其核心是事件循环，在其之上提供低级实用程序，例如：流抽象、异步 DNS 解析器、网络客户端/服务器、HTTP 客户端/服务器以及与进程的交互。
安装ReactPHP很简单，直接用composer安装

composer require react/react

ReactPHP 有各种组件，例如事件循环、promise 和流。当我们安装 ReactPHP 时，这些组件和其他一些组件都会被安装，因此您不必单独安装
ReactPHP异步编程的写法采用了类Promise的设计，熟悉ES6 Promise的话上手很快。

require __DIR__ . '/vendor/autoload.php';

$browser = new React\Http\Browser();
$url = 'https://www.baidu.com/?i=';
$start = microtime(true);

for ($i=0; $i < 500; $i++) { 
  $browser->get($url.$i)->then(function (Psr\Http\Message\ResponseInterface $response) use($i,$start){
    $result = $response->getBody();
    echo "第{$i}次访问百度完成，用时：".microtime(true)-$start."\n";
  }, function (Exception $e) {
    echo 'Error: ' . $e->getMessage() . PHP_EOL;
  });
}

上面的代码确实实现了异步编程。但我们发现用时并没有多大的提升

总用时为0.63秒？为什么？
我们把5次访问改为500次，运行程序时，我们新开个命令行窗口，输入ps -ef | grep php来观察php进程

我们发现始终都只有一个php ReactPhp2.php的进程，说明Reactphp没有采用多进程机制，但它的优点大于它的缺点。

1. ReactPHP安装非常方便，直接用composer require就行，不需要其他php扩展的支持
2. ReactPHP对异步编程进行了非常好的支持，书写语法非常友好。具体查看官方文档就知道了

Fibers

2021 年 11 月发布的 PHP 8.1 开始支持了一个新特性： Fibers^[3]，用于实现轻量级协程。由于它是一个非常底层的 API ，并不是直接可以使用于应用层，更多的，要使用Fibers，只需把php版本更新到8.1即可。

<?php
declare(ticks=1);//Zend引擎每执行1条低级语句就去执行一次 register_tick_function() 注册的函数

class Thread {
  protected static $names = [];
  protected static $fibers = [];
  protected static $params = [];

  public static function register(string|int $name, callable $callback, array $params)
  {
    self::$names[]  = $name;
    self::$fibers[] = new Fiber($callback);
    self::$params[] = $params;
  }

  public static function run() {
    $output = [];

    while (self::$fibers) {
      foreach (self::$fibers as $i => $fiber) {
          try {
              if (!$fiber->isStarted()) {
                  // Register a new tick function for scheduling this fiber
                  register_tick_function('Thread::scheduler');
                  $fiber->start(...self::$params[$i]);
              } elseif ($fiber->isTerminated()) {
                  $output[self::$names[$i]] = $fiber->getReturn();
                  unset(self::$fibers[$i]);
              } elseif ($fiber->isSuspended()) {
                $fiber->resume();
              }                
          } catch (Throwable $e) {
              $output[self::$names[$i]] = $e;
          }
      }
    }

    return $output;
  }

  public static function scheduler () {
    if(Fiber::getCurrent() === null) {
      return;
    }
    // running Fiber::suspend() in this if condition will prevent an infinite loop!
    if(count(self::$fibers) > 1)
    {
      Fiber::suspend();
    }
  }
}


$start = microtime(true);
function curl($url)
{
    $ch = curl_init();
    curl_setopt($ch, CURLOPT_URL, $url);
    curl_setopt($ch, CURLOPT_RETURNTRANSFER, 1);
    curl_setopt($ch, CURLOPT_HEADER, 0);
    $output = curl_exec($ch);
    curl_close($ch);
    return $output;
}
function callback($i){
    $url = 'https://www.baidu.com/?i=';
    return curl($url.$i);
};
// registering 6 Threads (A, B, C, D, E, and F)
foreach(range(1, 5) as $id) {
  Thread::register($id, function($id,$start){
    $result = callback($id);
    return microtime(true) - $start;//放回用时。
  }, [$id,$start]);
}

// run threads and wait until execution finishes
$outputs = Thread::run();


print_r($outputs);

以上代码通过declare(ticks=1)来实现Fibers协程的切换。通过register_tick_function函数来让程序没运行一行就执行Fiber::suspend()，这样就可以达到协程切换的目的，最后$outputs打印的时每个协程的耗时。也因为频繁suspend，耗时比较长。但我们异步编程的目录已达到了。

Amphp

前面的例子只是使用Fiber完成了一个最简单但不太合理的协程。实际上 Fiber 扩展进入内核后，它是一个非常底层的 API ，并不是直接用于应用层的。而Amphp^[4]利用了Fibers的特殊，完成一个完整的协程框架，里面包含了一大堆组件，可以根据需要composer对应的组件就行。
安装：

composer require amphp/amp

Amphp的使用和Reactphp很像，使用方法大概如下：

<?php

use Amp\Future;

$httpClient = HttpClientBuilder::buildDefault();
$uris = [
    "google" => "https://www.google.com",
    "news"   => "https://news.google.com",
    "bing"   => "https://www.bing.com",
    "yahoo"  => "https://www.yahoo.com",
];

try {
    $responses = Future\await(array_map(function ($uri) use ($httpClient) {
        return Amp\async(fn () => $httpClient->request(new Request($uri, 'HEAD')));
    }, $uris));

    foreach ($responses as $key => $response) {
        printf(
            "%s | HTTP/%s %d %s\n",
            $key,
            $response->getProtocolVersion(),
            $response->getStatus(),
            $response->getReason()
        );
    }
} catch (Exception $e) {
    // If any one of the requests fails the combo will fail
    echo $e->getMessage(), "\n";
}

Swoole

我们最后介绍Swoole，swoole目前应该是php中异步编程的首选框架了，在国内有很高的知名度，甚至在有些php招聘中要求必会swoole。

use Swoole\Coroutine\Channel;
use function Swoole\Coroutine\go;
use function Swoole\Coroutine\run;
run(function() {
    $chan = new Channel(5);

    go(function () use ($chan) {
        $cli = new Swoole\Coroutine\Http\Client('www.qq.com', 80);
        $ret = $cli->get('/');
        $chan->push(['key' => 'www.qq.com', 'content' => '访问www.qq.com成功！']);
    });

    go(function () use ($chan) {
        $cli = new Swoole\Coroutine\Http\Client('www.163.com', 80);
        $ret = $cli->get('/');
        $chan->push(['key' => 'www.163.com', 'content' => '访问www.qq.com成功！']);
    });

    for ($i = 0; $i < 2; $i++) {
        $result = $chan->pop();
        var_dump($result);
    }
});

上面的代码执行结果如下：

点赞(0) 微信打赏支付宝打赏

本文分类：PHP
浏览次数：39 次浏览
发布日期：2023-10-10 13:39:36
本文链接：https://www.889sw.com/show/NzIy

上一篇 > 简单了解一下CGI、FastCGI、PHP-FPM的区别
下一篇 > 如何发布一个PHP包到Composer

PHP异步编程探索

什么是异步编程

Php pcntl扩展

使用curl_multi函数组

Parallel扩展

ReactPHP

Fibers

Amphp

Swoole

评论列表共有 0 条评论

发表评论

PHP异步编程探索

什么是异步编程

Php pcntl扩展

使用curl_multi函数组

Parallel扩展

ReactPHP

Fibers

Amphp

Swoole

PHP常量：让你的代码更稳定、可读、易维护

PHP函数库解析：array_unique()

如何简化 Laravel 网站设置管理

php函数对象编程指南在跨平台开发中的兼容性是什么？

评论列表 共有 0 条评论

发表评论 取消回复

评论列表共有 0 条评论

发表评论