本篇内容主要为大家提供的是PHP编程中尝试程序并发的几种方式总结。感兴趣的同学可以参考学习下，具体内容如下： 这里有新鲜出炉的PHP面向对象教程，程序狗速度看过来！ PHP开源脚本语言PHP（外文名: Hypertext Preprocessor，中文名：“超文本预处理器”）是一种通用开源脚本语言
本篇内容主要为大家提供的是PHP编程中尝试程序并发的几种方式总结。感兴趣的同学可以参考学习下，具体内容如下：
这里有新鲜出炉的PHP面向对象教程，程序狗速度看过来！
PHP开源脚本语言
PHP（外文名: Hypertext Preprocessor，中文名：“超文本预处理器”）是一种通用开源脚本语言。语法吸收了C语言、Java和Perl的特点，入门门槛较低，易于学习，使用广泛，主要适用于Web开发领域。PHP的文件后缀名为php。 这篇文章主要介绍了PHP编程中尝试程序并发的几种方式总结,这里举了借助yield的异步以及swoole_process的进程创建等例子,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.phperz.com/');
$ch_2 = curl_init('http://www.phperz.com/');
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.phperz.comrn
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.phperz.com';
}
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亦可，仅为展示之用。
声明：本文内容来源自网络，文字、图片等素材版权属于原作者，平台转载素材出于传递更多信息，文章内容仅供参考与学习，切勿作为商业目的使用。如果侵害了您的合法权益，请您及时与我们联系，我们会在第一时间进行处理！我们尊重版权，也致力于保护版权，站搜网感谢您的分享！
