php多进程-100万个数的排序

100万个数的排序，使用冒泡法显然是会挂的，于是采用分片的方式分给10给进程去各自排序，然后汇总，汇总需要用到php进程间通讯之官道。

先执行 php input.php
再执行 php sort.php

生产数据
input.php

<?php
/**
+----------------------------------------------------------
* date: 2019-04-19 09:32:09
+----------------------------------------------------------
* author: Raoxiaoya
+----------------------------------------------------------
* describe: 生产100万个整数
+----------------------------------------------------------
*/
$file = './data.txt';
if(file_exists($file)){
unlink($file);
}

touch($file);

for ($i = 0; $i < 1000000; $i++) {
$num = rand(1,10000000);
file_put_contents($file, $num . ',', FILE_APPEND);
}

echo 'ok'.PHP_EOL;

--------------------------------------------------------------------------------------

执行排序

sort.php

<?php
/**
+----------------------------------------------------------
* date: 2019-04-19 09:38:03
+----------------------------------------------------------
* author: Raoxiaoya
+----------------------------------------------------------
* describe: 100万个整数排序
+----------------------------------------------------------
*/

$file = './data.txt';
$result_file = './sort_result.txt';
$n = 10;
$count = 1000000;

$start = microtime(true);

$cont = file_get_contents($file);

$arr = explode(',', $cont);
$arr = array_filter($arr);

// 找出最大值和最小值
$min = null;
$max = null;
foreach ($arr as $v) {
if(is_null($max) && is_null($min)){
$max = $v;
$min = $v;
}

if($v > $max){
$max = $v;
continue;
}

if($v < $min){
$min = $v;
continue;
}
}

// 分10片
$x = (int)(($max - $min) / $count) * ($count / $n);
$temp = [];
for($i = 0; $i < $n; $i++){
if($i == 0){
$temp[$i] = [$min, $min + $x];
}elseif($i == 9){
$temp[$i] = [$temp[$i - 1][1] + 1, $max];
}else{
$temp[$i] = [$temp[$i - 1][1] + 1, $temp[$i - 1][1] + 1 + $x];
}
}

$tempdata = [];
foreach ($temp as $key => $value) {
$tempdata[$key] = [];
foreach ($arr as $val) {
if($val >= $value[0] && $val <= $value[1]){
array_push($tempdata[$key], $val);
continue;
}
}
}

function sort_data($data){
$len = count($data);
for ($i = 0; $i < $len - 1; $i++) {
for ($j = 0; $j < $len - $i - 1; $j++) {
if($data[$j] > $data[$j+1]){
$temp = $data[$j];
$data[$j] = $data[$j+1];
$data[$j+1] = $temp;
}
}
}
return $data;
}

// 创建管道
$sPipePath = "my_pipe.".posix_getpid();
if (!posix_mkfifo($sPipePath, 0666)) {
die("create pipe {$sPipePath} error");
}
$childs = array();
// 启动十个进程
for ($i = 0; $i < $n; $i++) {
$pid = pcntl_fork();
if ($pid == -1)
die('Could not fork');
if ($pid) {
$childs[] = $pid;
} else {
$result = sort_data($tempdata[$i]);
file_put_contents($sPipePath, $i . '-' . json_encode($result) . PHP_EOL);
exit();
}
}

// 父进程
$oR = fopen($sPipePath, 'r');
stream_set_blocking($oR, FALSE); // 将管道设置为非堵塞，用于适应超时机制
$sData = ''; // 存放管道中的数据
$nLine = 0;
$nStart = time();
while ($nLine < $n) {
$sLine = fread($oR, 1024);
if (empty($sLine)) {
continue;
}
// 用于分析多少任务处理完毕，通过‘\n’标识
foreach(str_split($sLine) as $c) {
if ("\n" == $c) {
++$nLine;
}
}
$sData .= $sLine;
}
echo "Final line count:$nLine\n";
fclose($oR);
unlink($sPipePath);

// 等待子进程执行完毕，避免僵尸进程
while (count($childs) > 0) {
foreach ($childs as $key => $pid) {
$res = pcntl_waitpid($pid, $status, WNOHANG);
if ($res == -1 || $res > 0)
unset($childs[$key]);
}

sleep(1);
}

// 处理 $sData
$res = [];
foreach (explode("\n", $sData) as $item) {
if(!empty($item)){
$a = explode('-', $item);
$res[$a[0]] = json_decode($a[1], true);
}
}

ksort($res);
file_put_contents($result_file, json_encode(array_values($res)));

echo 'ok'.PHP_EOL;

$end = microtime(true);

echo 'time ' . ($end - $start) . PHP_EOL;

------------------------------------------------------------------------------

新开一个窗口
ps -ef |grep php

top

由于本机是虚拟机，cpu已使用100%