PHP编程中尝试程序并发的几种方式总结

本文大约总结了PHP编程中的五种并发方式：

1.curl_multi_init

文档中说的是 Allows the processing of multiple cURL handles asynchronously. 确实是异步。这里需要理解的是select这个方法，文档中是这么解释的Blocks until there is activity on any of the curl_multi connections.。了解一下常见的异步模型就应该能理解，select, epoll，都很有名

<?php // build the individual requests as above, but do not execute them $ch_1 = curl_init('http://www.jb51.net/'); $ch_2 = curl_init('http://www.jb51.net/'); curl_setopt($ch_1, CURLOPT_RETURNTRANSFER, true); curl_setopt($ch_2, CURLOPT_RETURNTRANSFER, true); // build the multi-curl handle, adding both $ch $mh = curl_multi_init(); curl_multi_add_handle($mh, $ch_1); curl_multi_add_handle($mh, $ch_2); // execute all queries simultaneously, and continue when all are complete $running = null; do { curl_multi_exec($mh, $running); $ch = curl_multi_select($mh); if($ch !== 0){ $info = curl_multi_info_read($mh); if($info){ var_dump($info); $response_1 = curl_multi_getcontent($info['handle']); echo "$response_1 n"; break; } } } while ($running > 0); //close the handles curl_multi_remove_handle($mh, $ch_1); curl_multi_remove_handle($mh, $ch_2); curl_multi_close($mh);

这里我设置的是，select得到结果，就退出循环，并且删除 curl resource, 从而达到取消http请求的目的。

2.swoole_client

swoole_client提供了异步模式，我竟然把这个忘了。这里的sleep方法需要swoole版本大于等于1.7.21, 我还没升到这个版本，所以直接exit也可以。

<?php $client = new swoole_client(SWOOLE_SOCK_TCP, SWOOLE_SOCK_ASYNC); //设置事件回调函数 $client->on("connect", function($cli) { $req = "GET / HTTP/1.1rn Host: www.jb51.netrn Connection: keep-alivern Cache-Control: no-cachern Pragma: no-cachernrn"; for ($i=0; $i < 3; $i++) { $cli->send($req); } }); $client->on("receive", function($cli, $data){ echo "Received: ".$data."n"; exit(0); $cli->sleep(); // swoole >= 1.7.21 }); $client->on("error", function($cli){ echo "Connect failedn"; }); $client->on("close", function($cli){ echo "Connection closen"; }); //发起网络连接 $client->connect('183.207.95.145', 80, 1);

3.process

哎，竟然差点忘了 swoole_process, 这里就不用 pcntl 模块了。但是写完发现，这其实也不算是中断请求，而是哪个先到读哪个，忽视后面的返回值。

<?php $workers = []; $worker_num = 3;//创建的进程数 $finished = false; $lock = new swoole_lock(SWOOLE_MUTEX); for($i=0;$i<$worker_num ; $i++){ $process = new swoole_process('process'); //$process->useQueue(); $pid = $process->start(); $workers[$pid] = $process; } foreach($workers as $pid => $process){ //子进程也会包含此事件 swoole_event_add($process->pipe, function ($pipe) use($process, $lock, &$finished) { $lock->lock(); if(!$finished){ $finished = true; $data = $process->read(); echo "RECV: " . $data.PHP_EOL; } $lock->unlock(); }); } function process(swoole_process $process){ $response = 'http response'; $process->write($response); echo $process->pid,"t",$process->callback .PHP_EOL; } for($i = 0; $i < $worker_num; $i++) { $ret = swoole_process::wait(); $pid = $ret['pid']; echo "Worker Exit, PID=".$pid.PHP_EOL; }

4.pthreads

编译pthreads模块时，提示php编译时必须打开ZTS, 所以貌似必须 thread safe 版本才能使用. wamp中多php正好是TS的，直接下了个dll, 文档中的说明复制到对应目录，就在win下测试了。还没完全理解，查到文章说 php 的 pthreads 和 POSIX pthreads是完全不一样的。代码有些烂，还需要多看看文档，体会一下。

<?php class Foo extends Stackable { public $url; public $response = null; public function __construct(){ $this->url = 'http://www.jb51.net'; } public function run(){} } class Process extends Worker { private $text = ""; public function __construct($text,$object){ $this->text = $text; $this->object = $object; } public function run(){ while (is_null($this->object->response)){ print " Thread {$this->text} is runningn"; $this->object->response = 'http response'; sleep(1); } } } $foo = new Foo(); $a = new Process("A",$foo); $a->start(); $b = new Process("B",$foo); $b->start(); echo $foo->response;

5.yield

以同步方式书写异步代码：

<?php class AsyncServer { protected $handler; protected $socket; protected $tasks = []; protected $timers = []; public function __construct(callable $handler) { $this->handler = $handler; $this->socket = socket_create(AF_INET, SOCK_DGRAM, SOL_UDP); if(!$this->socket) { die(socket_strerror(socket_last_error())."n"); } if (!socket_set_nonblock($this->socket)) { die(socket_strerror(socket_last_error())."n"); } if(!socket_bind($this->socket, "0.0.0.0", 1234)) { die(socket_strerror(socket_last_error())."n"); } } public function Run() { while (true) { $now = microtime(true) * 1000; foreach ($this->timers as $time => $sockets) { if ($time > $now) break; foreach ($sockets as $one) { list($socket, $coroutine) = $this->tasks[$one]; unset($this->tasks[$one]); socket_close($socket); $coroutine->throw(new Exception("Timeout")); } unset($this->timers[$time]); } $reads = array($this->socket); foreach ($this->tasks as list($socket)) { $reads[] = $socket; } $writes = NULL; $excepts= NULL; if (!socket_select($reads, $writes, $excepts, 0, 1000)) { continue; } foreach ($reads as $one) { $len = socket_recvfrom($one, $data, 65535, 0, $ip, $port); if (!$len) { //echo "socket_recvfrom fail.n"; continue; } if ($one == $this->socket) { //echo "[Run]request recvfrom succ. data=$data ip=$ip port=$portn"; $handler = $this->handler; $coroutine = $handler($one, $data, $len, $ip, $port); if (!$coroutine) { //echo "[Run]everything is done.n"; continue; } $task = $coroutine->current(); //echo "[Run]AsyncTask recv. data=$task->data ip=$task->ip port=$task->port timeout=$task->timeoutn"; $socket = socket_create(AF_INET, SOCK_DGRAM, SOL_UDP); if(!$socket) { //echo socket_strerror(socket_last_error())."n"; $coroutine->throw(new Exception(socket_strerror(socket_last_error()), socket_last_error())); continue; } if (!socket_set_nonblock($socket)) { //echo socket_strerror(socket_last_error())."n"; $coroutine->throw(new Exception(socket_strerror(socket_last_error()), socket_last_error())); continue; } socket_sendto($socket, $task->data, $task->len, 0, $task->ip, $task->port); $deadline = $now + $task->timeout; $this->tasks[$socket] = [$socket, $coroutine, $deadline]; $this->timers[$deadline][$socket] = $socket; } else { //echo "[Run]response recvfrom succ. data=$data ip=$ip port=$portn"; list($socket, $coroutine, $deadline) = $this->tasks[$one]; unset($this->tasks[$one]); unset($this->timers[$deadline][$one]); socket_close($socket); $coroutine->send(array($data, $len)); } } } } } class AsyncTask { public $data; public $len; public $ip; public $port; public $timeout; public function __construct($data, $len, $ip, $port, $timeout) { $this->data = $data; $this->len = $len; $this->ip = $ip; $this->port = $port; $this->timeout = $timeout; } } function AsyncSendRecv($req_buf, $req_len, $ip, $port, $timeout) { return new AsyncTask($req_buf, $req_len, $ip, $port, $timeout); } function RequestHandler($socket, $req_buf, $req_len, $ip, $port) { //echo "[RequestHandler] before yield AsyncTask. REQ=$req_bufn"; try { list($rsp_buf, $rsp_len) = (yield AsyncSendRecv($req_buf, $req_len, "127.0.0.1", 2345, 3000)); } catch (Exception $ex) { $rsp_buf = $ex->getMessage(); $rsp_len = strlen($rsp_buf); //echo "[Exception]$rsp_bufn"; } //echo "[RequestHandler] after yield AsyncTask. RSP=$rsp_bufn"; socket_sendto($socket, $rsp_buf, $rsp_len, 0, $ip, $port); } $server = new AsyncServer(RequestHandler); $server->Run(); ?>

代码解读：

借助PHP内置array能力，实现简单的“超时管理”，以毫秒为精度作为时间分片；

封装AsyncSendRecv接口，调用形如yield AsyncSendRecv()，更加自然；

添加Exception作为错误处理机制，添加ret_code亦可，仅为展示之用。