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 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;"> on("connect",function($cli) { $req = "GET / HTTP/1.1\r\n Host: www.jb51.cc\r\n Connection: keep-alive\r\n Cache-Control: no-cache\r\n Pragma: no-cache\r\n\r\n";

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 Failed\n";
});
$client->on("close",function($cli){
echo "Connection close\n";
});
//发起网络连接
$client->connect('183.207.95.145',80,1);

3.process

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

PHP;"> 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;"> url = '//www.jb51.cc'; } 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 running\n";
$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,1000)) { 
    continue; 
  } 

  foreach ($reads as $one) { 
    $len = socket_recvfrom($one,$data,65535,$ip,$port); 
    if (!$len) { 
      //echo "socket_recvfrom fail.\n"; 
      continue; 
    } 
    if ($one == $this->socket) { 
      //echo "[Run]request recvfrom succ. data=$data ip=$ip port=$port\n"; 
      $handler = $this->handler; 
      $coroutine = $handler($one,$len,$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->timeout\n"; 
      $socket = socket_create(AF_INET,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,$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=$port\n"; 
      list($socket,$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,$port,$timeout) {
$this->data = $data;
$this->len = $len;
$this->ip = $ip;
$this->port = $port;
$this->timeout = $timeout;
}
}

function AsyncSendRecv($req_buf,$req_len,$timeout) {
return new AsyncTask($req_buf,$timeout);
}

$server = new AsyncServer(RequestHandler);
$server->Run();

代码解读：

借助PHP内置array能力，实现简单的“超时管理”，以毫秒为精度作为时间分片；封装AsyncSendRecv接口，调用形如yield AsyncSendRecv()，更加自然；添加Exception作为错误处理机制，添加ret_code亦可，仅为展示之用。

原文链接：https://www.f2er.com/php/20145.html