PHP編程中嘗試程序并發(fā)的幾種方式總結(jié)

2020-03-22 19:45:07

字體：大中小

供稿：網(wǎng)友

1.curl_multi_init
文檔中說的是 Allows the processing of multiple cURL handles asynchronously. 確實(shí)是異步。這里需要理解的是select這個(gè)方法，文檔中是這么解釋的Blocks until there is activity on any of the curl_multi connections.。了解一下常見的異步模型就應(yīng)該能理解，select, epoll，都很有名
// build the individual requests as above, but do not execute them$ch_1 = curl_init('http://www.phpstudy.net/');$ch_2 = curl_init('http://www.phpstudy.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; 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 //close the handlescurl_multi_remove_handle($mh, $ch_1);curl_multi_remove_handle($mh, $ch_2);curl_multi_close($mh);這里我設(shè)置的是，select得到結(jié)果，就退出循環(huán)，并且刪除 curl resource, 從而達(dá)到取消http請求的目的。2.swoole_client
swoole_client提供了異步模式，我竟然把這個(gè)忘了。這里的sleep方法需要swoole版本大于等于1.7.21, 我還沒升到這個(gè)版本，所以直接exit也可以。
$client = new swoole_client(SWOOLE_SOCK_TCP, SWOOLE_SOCK_ASYNC);//設(shè)置事件回調(diào)函數(shù)$client- on("connect", function($cli) { $req = "GET / HTTP/1.1/r/n Host: www.phpstudy.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 $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";//發(fā)起網(wǎng)絡(luò)連接$client- connect('183.207.95.145', 80, 1);3.process
哎，竟然差點(diǎn)忘了 swoole_process, 這里就不用 pcntl 模塊了。但是寫完發(fā)現(xiàn)，這其實(shí)也不算是中斷請求，而是哪個(gè)先到讀哪個(gè)，忽視后面的返回值。
$workers = [];$worker_num = 3;//創(chuàng)建的進(jìn)程數(shù)$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){ //子進(jìn)程也會(huì)包含此事件 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模塊時(shí)，提示php編譯時(shí)必須打開ZTS, 所以貌似必須 thread safe 版本才能使用. wamp中多php正好是TS的，直接下了個(gè)dll, 文檔中的說明復(fù)制到對應(yīng)目錄，就在win下測試了。還沒完全理解，查到文章說 php 的 pthreads 和 POSIX pthreads是完全不一樣的。代碼有些爛，還需要多看看文檔，體會(huì)一下。
html' target='_blank'>class Foo extends Stackable { public $url; public $response = null; public function __construct(){ $this- url = 'http://www.phpstudy.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 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, 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內(nèi)置array能力，實(shí)現(xiàn)簡單的“超時(shí)管理”，以毫秒為精度作為時(shí)間分片；
封裝AsyncSendRecv接口，調(diào)用形如yield AsyncSendRecv()，更加自然；
添加Exception作為錯(cuò)誤處理機(jī)制，添加ret_code亦可，僅為展示之用。PHP教程

鄭重聲明：本文版權(quán)歸原作者所有，轉(zhuǎn)載文章僅為傳播更多信息之目的，如作者信息標(biāo)記有誤，請第一時(shí)間聯(lián)系我們修改或刪除，多謝。

上一篇：laravel-admin集成ueditor編輯器的圖片列表顯示問題解

下一篇：php新浪通行證新浪微博模擬統(tǒng)一登錄(后臺網(wǎng)頁抓