May 11, 2021 PHP
1. First, php simulation implements three methods of multithreading
2. Second, the real implementation of php multithreaded approach
When we were working on the project, some of the requirements, especially the response processing of the data, took a lot of time, because php is a short life cycle scripting language, to the default 30 seconds, php data processing is not complete, php life cycle is over. A t this point, you need to use asynchronous processing policy, that is, a php call can make multiple requests, these requests are not executed in order, but can be executed asynchronously, some requests are used to process data in the background, some requests are used to accept background response state, according to the state, with the user to do some simple interaction. But here's the problem, we all know that php itself doesn't support multithreaded, so how do we implement php multithreaded?
1, php multithreaded under linux
What is said below is a function of pcntl_fork php. B
ecause this function relies on the implementation of the operating system fork, the things described in this article apply only to linux/unix.
So take a look at how this function is used.php it says in the manual:
<?php
$pid = pcntl_fork();
if ($pid == -1) {
die('could not fork');
} else if ($pid) {
// we are the parent
pcntl_wait($status); //Protect against Zombie children
} else {
// we are the child
}
?>
By pcntl_fork a sub-process, if the return value is -1, then the sub-process creation fails. C reating a successful process id returns to the parent process and 0 to the child process. It's hard to understand, so it should be written like this:
<?php
$pid = pcntl_fork();
if($pid == -1){
//创建失败咱就退出呗,没啥好说的
die('could not fork');
}
else{
if($pid){
//从这里开始写的代码是父进程的,因为写的是系统程序,记得退出的时候给个返回值
exit(0);
}
else{
//从这里开始写的代码都是在新的进程里执行的,同样正常退出的话,最好也给一个返回值
exit(0);
}
}
?>
This change is much more understandable, if your parent process wants to know that the child process exits normally, you can add the pcntl_wait.
2. Stream_socket_client methods
function sendStream() {
$english_format_number = number_format($number, 4, '.', '');
echo $english_format_number;
exit();
$timeout = 10;
$result = array();
$sockets = array();
$convenient_read_block = 8192;
$host = "test.local.com";
$sql = "select waybill_id,order_id from xm_waybill where status>40 order by update_time desc limit 1 ";
$data = Yii::app()->db->createCommand($sql)->queryAll();
$id = 0;
foreach ($data as $k => $v) {
if ($k % 2 == 0) {
$send_data[$k]['body'] = NoticeOrder::getSendData($v['waybill_id']);
} else {
$send_data[$k]['body'] = array($v['order_id'] => array('extra' => 16));
}
$data = json_encode($send_data[$k]['body']);
$s = stream_socket_client($host . ":80", $errno, $errstr, $timeout, STREAM_CLIENT_ASYNC_CONNECT | STREAM_CLIENT_CONNECT);
if ($s) {
$sockets[$id++] = $s;
$http_message = "GET /php/test.php?data=" . $data . " HTTP/1.0\r\nHost:" . $host . "\r\n\r\n";
fwrite($s, $http_message);
} else {
echo "Stream " . $id . " failed to open correctly.";
}
}
while (count($sockets)) {
$read = $sockets;
stream_select($read, $w = null, $e = null, $timeout);
if (count($read)) {
/* stream_select generally shuffles $read, so we need to
compute from which socket(s) we're reading. */
foreach ($read as $r) {
$id = array_search($r, $sockets);
$data = fread($r, $convenient_read_block);
if (strlen($data) == 0) {
echo "Stream " . $id . " closes at " . date('h:i:s') . ".<br> ";
fclose($r);
unset($sockets[$id]);
} else {
$result[$id] = $data;
}
}
} else {
/* A time-out means that *all* streams have failed
to receive a response. */
echo "Time-out!\n";
break;
}
}
print_r($result);
}
3, through multi-process instead of multi-threaded
function daemon($func_name,$args,$number){
while(true){
$pid=pcntl_fork();
if($pid==-1){
echo "fork process fail";
exit();
}elseif($pid){//创建的子进程
static $num=0;
$num++;
if($num>=$number){
//当进程数量达到一定数量时候,就对子进程进行回收。
pcntl_wait($status);
$num--;
}
}else{ //为0 则代表是子进程创建的,则直接进入工作状态
if(function_exists($func_name)){
while (true) {
$ppid=posix_getpid();
var_dump($ppid);
call_user_func_array($func_name,$args);
sleep(2);
}
}else{
echo "function is not exists";
}
exit();
}
}
}
function worker($args){
//do something
}
daemon('worker',array(1),2);
The next page finds version 2
Download it, this v2 is only available for php5
Download it down, install:
Alternatively, you download it directly like this:
cd /tools
wget https://github.com/krakjoe/pthreads/archive/v2.0.10.zip
unzip v2.0.10.zip
cd pthreads-2.0.10
/usr/local/php/bin/phpize
./configure --with-php-config=/usr/local/php/bin/php-config
make
make install
Note: Your php needs to be turned on at compile time - enable-maintainer-zts
./configure --prefix=/usr/local/php --disable-fileinfo --enable-fpm --with-config-file-path=/etc --with-config-file-scan-dir=/etc/php.d --with-openssl --with-zlib --with-curl --enable-ftp --with-gd --with-xmlrpc --with-jpeg-dir --with-png-dir --with-freetype-dir --enable-gd-native-ttf --enable-mbstring --with-mcrypt=/usr/local/libmcrypt --enable-zip --with-mysql=/usr/local/mysql --without-pear --enable-maintainer-zts
vim /etc/php.ini
添加
extension=pthreads.so
重启php
/etc/init.d/php-fpm restart