基于 swoole 的压测工具

<?php
require_once(__DIR__ . '/BenchmarkApplicationImplV1.php');
use Swoole\Coroutine;
//swoole hook curl
Swoole\Coroutine::set(['hook_flags' => SWOOLE_HOOK_ALL | SWOOLE_HOOK_CURL]);
$totalRequest = 30000;
$concurrent = 500;
$startTime = microtime(true);
var_dump("起始时间：" . $startTime . "; 并发数：" . $concurrent . "; 请求数：" . $totalRequest);
$application = new BenchmarkApplicationImplV1('http://www.baidu.com', $totalRequest, $concurrent);
$application->start();
$endTime = microtime(true);
var_dump("结束时间：" . $endTime . "，耗时：" . ($endTime - $startTime));

<?php
/**
 * 主调度类接口
 */
interface BenchmarkApplication {
    /**
     * 启动请求。
     */
    public function start();
}

<?php
require_once(__DIR__ . '/BenchmarkApplication.php');
require_once(__DIR__ . '/Request.php');
require_once(__DIR__ . '/TaskUtilImplV1.php');
use Swoole\Process;
use Swoole\Coroutine;
use Swoole\Coroutine\WaitGroup;
/**
 * 主进程实现类
 */
class BenchmarkApplicationImplV1 implements BenchmarkApplication
{
    private $request;
    private $totalRequest;
    private $concurrent;
    private $cpuNum;
    /**
     * 结果数组
     */
    private $resultArr;
    private $resultEnd = false;
    /**
     * 运行时进程数
     *
     * [
     * [
     * 'total_request' => 100,
     * 'concurrent' => 10,
     * ],
     * [
     * 'total_request' => 100,
     * 'concurrent' => 11,
     * ]
     * ]
     */
    private $runParam;
    /**
     * 默认单个进程可以处理的并发数
     */
    private $defaultConcurentOneProcess = 30;
    /**
     * 初始化启动。
     * 接收url，请求总数，并发数。
     * 返回一个当前类的对象。
     * 根据cpu个数，当前请求数，计算应启动的进程数，以及每个进程需要处理的请求个数
     */
    public function __construct(string $url, int $totalRequest, int $concurrent)
    {
        $this->request = new Request();
        $this->request->setUrl($url);
        $this->cpuNum = $this->getCpuNum();
        $this->totalRequest = $totalRequest;
        $this->concurrent = $concurrent;
        $this->resultArr = [];
    }
    /**
     * 获取当前机器的cpu数量
     * @return
     */
    protected function getCpuNum(): int
    {
        return swoole_cpu_num();
    }
    /**
     * 启动请求。
     */
    public function start()
    {
        $this->init();
        $this->runProcess();
    }
    public function getResult()
    {
        return $this->resultArr;
    }
    /**
     * 根据并发数、总请求数、cpu数
     * 计算应启动进程数、每个进程分配的请求数、每个进程的并发数
     */
    protected function init()
    {
        $this->runParam = [];
        //如果并发数超过 cpu数量 * 默认进程处理的并发数
        //那么进程数 = cpu数量
        //否则 单进程并发数 = 总并发数 / 进程数
        if (ceil($this->concurrent / $this->defaultConcurentOneProcess) >= $this->cpuNum) {
            $processNum = $this->cpuNum;
        } else {
            $processNum = ceil($this->concurrent / $this->defaultConcurentOneProcess);
        }
        //根据进程数计算平均请求数和平均并发数
        //除不尽的部分归到最后一个进程上
        $aveRequest = floor($this->totalRequest / $processNum);
        $aveConcurremt = floor($this->concurrent / $processNum);
        while ($processNum--) {
            if ($this->totalRequest <= 0) {
                break;
            }
            $tmp = [];
            $tmp['total_request'] = $aveRequest;
            $tmp['concurrent'] = $aveConcurremt;
            $this->totalRequest -= $aveRequest;
            $this->concurrent -= $aveConcurremt;
            if ($this->totalRequest < $aveRequest) {
                $tmp['total_request'] += $this->totalRequest;
                $this->totalRequest = 0;
            }
            if ($this->concurrent < $aveConcurremt) {
                $tmp['concurrent'] += $this->concurrent;
                $this->concurrent = 0;
            }
            $this->runParam[] = $tmp;
        }
    }
    /**
     * 执行进程
     */
    protected function runProcess()
    {
        $process = [];
        //创建进程
        foreach ($this->runParam as $oneProcessParam) {
            //延迟1S启动，对齐启动时间
            $timeStart = time() + 1;
            $process[] = new Process(function (Process $proc) use ($oneProcessParam, $timeStart) {
                $socket = $proc->exportSocket();
                $taskUtil = new TaskUtilImplV1($this->request, $oneProcessParam['total_request'], $oneProcessParam['concurrent']);
                while (time() < $timeStart) {
                    //阻塞等待，时间对齐
                }
                $taskUtil->doTask();
                $result = $taskUtil->getResult();
                foreach ($result as $one) {
                    $socket->send($one);
                }
                $socket->close();
                $proc->exit();
            }, false, 2, 1);
        }
        //启动进程
        $sockets = [];
        foreach ($process as $oneProcess) {
            $oneProcess->start();
            $sockets[] = $oneProcess->exportSocket();
        }
        //主进程开多个协程，等待子进程返回数据
        $this->resultEnd = false;
        Coroutine::create(function () use ($sockets) {
            $wg = new WaitGroup();
            foreach ($sockets as $oneSocket) {
                $wg->add();
                Coroutine::create(function () use ($wg, $oneSocket) {
                    $countResult = 0;
                    while ($oneSocket->checkLiveness()) {
                        $oneResult = $oneSocket->recv(65535, 1);
                        if (!empty($oneResult)) {
                            $countResult++;
                            $this->resultArr[] = $oneResult;
                        }
                    }
                    $wg->done();
                });
            }
            $wg->wait();
            //打印结果
            $this->dumpResult();
        });
        foreach ($process as $oneProcess) {
            $oneProcess->wait(true);
        }
        Process::wait(true);
    }
    /**
     * 输出结果
     */
    private function dumpResult() {
        //输出结果
        $totalCost = array_sum($this->resultArr);
        var_dump("结果总数：". count($this->resultArr));
        var_dump("总耗时：". $totalCost. " s");
        var_dump("平均耗时：". bcdiv($totalCost, count($this->resultArr), 4). " s");
        sort($this->resultArr);
        //截取其中的90%
        $this->resultArr = array_slice($this->resultArr, 0, floor(count($this->resultArr) * 0.95) );
        var_dump("95% 数据");
        $totalCost = array_sum($this->resultArr);
        var_dump("结果总数：". count($this->resultArr));
        var_dump("总耗时：". $totalCost. " s");
        var_dump("平均耗时：". bcdiv($totalCost, count($this->resultArr), 4). " s");
    }
}

<?php
/**
 * 子进程处理单元接口类
 */
interface TaskUtil {
    /**
     * 执行任务
     */
    public function doTask();
}

<?php
require_once(__DIR__ . '/TaskUtil.php');
require_once(__DIR__ . '/Request.php');
use Swoole\Coroutine;
use Swoole\Coroutine\WaitGroup;
/**
 * 子进程处理单元
 
 * 请求任务执行单元
 * 根据并发数启动对应的协程执行请求
 */
class TaskUtilImplV1 implements TaskUtil
{
    /**
     * @var Request
     */
    private $request;
    /**
     * 请求总数
     */
    private $totalRequest;
    /**
     * 并发数
     */
    private $concurrent;
    private $result;
    /**
     *  初始化。
     * 接收请求任务对象，总请求数。
     * 返回当前类的实例。
     */
    public function __construct(Request $request, int $totalRequest, int $concurrent)
    {
        $this->request = $request;
        $this->totalRequest = $totalRequest;
        $this->concurrent = $concurrent;
        $this->resultRequest = [];
    }
    /**
     * 执行任务
     */
    public function doTask()
    {
        $totalRequest = $this->totalRequest;
        $concurrent = $this->concurrent;
        //启动 $concurrent 个协程处理请求，平均每个处理为 $totalRequest / $concurrent
        $aveRuest = floor($totalRequest / $concurrent);
        $wg = new WaitGroup();
        while ($concurrent--) {
            //计算当前协程应请求个数，除不尽的个数归到最后一个协程
            $requestNum = $aveRuest;
            $totalRequest -= $aveRuest;
            $totalRequest < $aveRuest && $requestNum += $totalRequest;
            //启动协程执行请求,每个协程启动
            $wg->add();
            Coroutine::create(function () use ($wg, $requestNum) {
                while ($requestNum--) {
                    $timeStart = microtime(true);
                    Request::request($this->request->getUrl());
                    //保存时间
                    $this->resultRequest[] = microtime(true) - $timeStart;
                }
                $wg->done();
            });
        }
        $wg->wait();
    }
    public function getResult()
    {
        return $this->resultRequest;
    }
}

<?php
/**
 *  请求url记录
 * 结合本次作业，做了简单处理
 */
class Request {
    private $url;
    /**
     * Default constructor
     */
    public function __construct() {
    }
    /**
     * 设置请求的参数
     */
    public function setUrl(string $url) {
        $this->url = $url;
    }
    public function getUrl() {
        return $this->url;
    }
    /**
    * 请求执行单元
     *结合本次作业，简单做了一个请求单元，如果实际使用，当前类需要重构
    */
    public static function request($url) {
        $headerArray =array("Content-type:application/json;","Accept:application/json");
        $ch = curl_init();
        curl_setopt($ch, CURLOPT_URL, $url);
        curl_setopt($ch, CURLOPT_SSL_VERIFYPEER, FALSE); 
        curl_setopt($ch, CURLOPT_SSL_VERIFYHOST, FALSE); 
        curl_setopt($ch, CURLOPT_RETURNTRANSFER, 1);
        curl_setopt($ch,CURLOPT_HTTPHEADER,$headerArray);
        $output = curl_exec($ch);
        curl_close($ch);
        $output = json_decode($output,true);
        return $output;
    }
}