PHP編程中嘗試程序并發(fā)的幾種方式總結(jié)
更新時間：2016年03月21日 14:43:31 作者：一堆好人卡
這篇文章主要介紹了PHP編程中嘗試程序并發(fā)的幾種方式總結(jié),這里舉了借助yield的異步以及swoole_process的進(jìn)程創(chuàng)建等例子,PHP本身并不支持多線程并發(fā),需要的朋友可以參考下
本文大約總結(jié)了PHP編程中的五種并發(fā)方式：
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.。了解一下常見的異步模型就應(yīng)該能理解，select, epoll，都很有名
<?php
// build the individual requests as above, but do not execute them
$ch_1 = curl_init('http://www.dbjr.com.cn/');
$ch_2 = curl_init('http://www.dbjr.com.cn/');
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);
這里我設(shè)置的是，select得到結(jié)果，就退出循環(huán)，并且刪除 curl resource, 從而達(dá)到取消http請求的目的。
2.swoole_client
swoole_client提供了異步模式，我竟然把這個忘了。這里的sleep方法需要swoole版本大于等于1.7.21, 我還沒升到這個版本，所以直接exit也可以。
<?php
$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.dbjr.com.cn\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";
});
//發(fā)起網(wǎng)絡(luò)連接
$client->connect('183.207.95.145', 80, 1);
3.process
哎，竟然差點忘了 swoole_process, 這里就不用 pcntl 模塊了。但是寫完發(fā)現(xiàn)，這其實也不算是中斷請求，而是哪個先到讀哪個，忽視后面的返回值。
<?php
$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)程也會包含此事件
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, 文檔中的說明復(fù)制到對應(yīng)目錄，就在win下測試了。 還沒完全理解，查到文章說 php 的 pthreads 和 POSIX pthreads是完全不一樣的。代碼有些爛，還需要多看看文檔，體會一下。
<?php
class Foo extends Stackable {
public $url;
public $response = null;
public function __construct(){
$this->url = 'http://www.dbjr.com.cn';
}
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能力，實現(xiàn)簡單的“超時管理”，以毫秒為精度作為時間分片；
封裝AsyncSendRecv接口，調(diào)用形如yield AsyncSendRecv()，更加自然；
添加Exception作為錯誤處理機(jī)制，添加ret_code亦可，僅為展示之用。
相關(guān)文章
最近在PHP項目中使用到了hmac_sha256加密方式,下面這篇文章主要給大家介紹了關(guān)于PHP實現(xiàn)sha-256哈希算法的相關(guān)資料,文中通過實例代碼介紹的非常詳細(xì),需要的朋友可以參考下2023-01-01
這篇文章主要介紹了php常用的url處理函數(shù),實例總結(jié)了parse_url、rawurldecode、urlencode、urldecode等一些較為常用的URL處理函數(shù),非常具有實用價值,需要的朋友可以參考下2014-11-11
本篇文章是對PHP中的curl參數(shù)進(jìn)行了詳細(xì)的分析介紹，需要的朋友參考下2013-06-06
這篇文章主要介紹了PHP遍歷數(shù)組的三種方法及效率對比,實例分析了foreach、while與for三種遍歷數(shù)組的方法與相關(guān)的效率比對,具有一定參考借鑒價值,需要的朋友可以參考下2015-02-02
php 使用file_get_contents讀取大文件的方法
本文介紹了在php中使用file_get_contents函數(shù)讀取大文件的方法，并附上了示例以及使用小技巧，非常的實用，這里推薦給大家2014-11-11
這篇文章主要介紹了php動態(tài)生成縮略圖并輸出顯示的方法,涉及php操作圖片的相關(guān)技巧,非常具有實用價值,需要的朋友可以參考下2015-04-04
這篇文章主要介紹了PHP引用的調(diào)用方法,結(jié)合實例形式分析了PHP中引用的常見調(diào)用技巧,需要的朋友可以參考下2016-04-04
這篇文章主要介紹了php簡單判斷兩個字符串是否相等的方法,涉及php針對字符串操作的相關(guān)技巧,具有一定參考借鑒價值,需要的朋友可以參考下2015-07-07
這篇文章主要介紹了Hbuilder中PHP環(huán)境配置，想用Hbuilder工具開發(fā)的同學(xué)，一定要仔細(xì)看一下2021-04-04
工廠模式包含普通工廠模式和抽象工廠模式，但是，不管是什么工廠模式，它們都是有一個作用，那就是生成對象2012-02-02