本文大约总结了php编程中的五种并发方式： 1.curl_multi_init 文档中说的是 allows the processing of multiple curl handles asynchronously. 确实是异步。这里需要理解的是select这个方法，文档中是这么解释的blocks until there is activity on any of th
本文大约总结了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，都很有名
0);//close the handlescurl_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也可以。
on(connect, function($cli) { $req = get / http/1.1\r\n host: www.jb51.net\r\n connection: keep-alive\r\n cache-control: no-cache\r\n pragma: no-cache\r\n\r\n; for ($i=0; $i 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 模块了。但是写完发现，这其实也不算是中断请求，而是哪个先到读哪个，忽视后面的返回值。
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 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
以同步方式书写异步代码：
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=$port\n; $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->timeout\n; $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=$port\n; 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_buf\n; 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_buf\n; } //echo [requesthandler] after yield asynctask. rsp=$rsp_buf\n; socket_sendto($socket, $rsp_buf, $rsp_len, 0, $ip, $port); } $server = new asyncserver(requesthandler); $server->run(); ?>
代码解读：
借助php内置array能力，实现简单的“超时管理”，以毫秒为精度作为时间分片；
封装asyncsendrecv接口，调用形如yield asyncsendrecv()，更加自然；
添加exception作为错误处理机制，添加ret_code亦可，仅为展示之用。