初试 php-swool

介绍

swool官网网站：swool
PHP的异步、并行、高性能网络通信引擎，使用纯C语言编写，提供了PHP语言的异步多线程服务器，异步TCP/UDP网络客户端，异步MySQL，异步Redis，数据库连接池，AsyncTask，消息队列，毫秒定时器，异步文件读写，异步DNS查询。
Swoole内置了Http/WebSocket服务器端/客户端、Http2.0服务器端。
Swoole可以广泛应用于互联网、移动通信、企业软件、云计算、网络游戏、物联网（IOT）、车联网、智能家居等领域。 使用PHP+Swoole作为网络通信框架，可以使企业IT研发团队的效率大大提升，更加专注于开发创新产品。

一个简单的http server

$serv = new Swoole\Http\Server("127.0.0.1", 9502);
$serv->on('Request', function($request, $response) {
    var_dump($request->get);
    var_dump($request->post);
    var_dump($request->cookie);
    var_dump($request->files);
    var_dump($request->header);
    var_dump($request->server);
    $response->cookie("User", "Swoole");
    $response->header("X-Server", "Swoole");
    $response->end("<h1>Hello Swoole!</h1>");
});
$serv->start();

使用中的修改

项目结构

+/config--------配置文件
    |/config_develop.php
    |/config_test.php
    |/config_product.php
    |/swool.lvphp.com.conf
+/controller----控制器
    |/IndexController.php
+/process-------进程
    |/Worker.php   worker进程
    |/Tasker.php   task进程
+/libs----------类库文件
    |/Application.php
    |/RedisDB.php
+/public--------静态文件
    |/index.html
    |/favicon.ico
+server.php-----启动文件

启动程序

server.php
<?php
define("ROOT_PATH", dirname(__FILE__));
define("LOG_NAME", "swool.lvphp.com");
define("LOG_PATH", "/data/logs/php/");
define("API_CALLER", "swool");
//note 加载配置文件
$server_type = get_cfg_var('yaf.environ');
$conf_file = ROOT_PATH . "/config/config_{$server_type}.php";
if(!is_file($conf_file)){
    exit("no config file");
}
require(ROOT_PATH . '/libs/Application.php');
Application::load(array("dir"=>array(ROOT_PATH . "/controller", ROOT_PATH . "/libs", ROOT_PATH . "/process"), "file"=>array("Config"=>$conf_file)));

class HttpServer{
    public static $instance;

    public function __construct($conf) {
        //note 开启服务
        $http = new swoole_http_server(Config::SWOOL_ADDR, Config::SWOOL_PORT);
        $http->set($conf);
        $http->on("start", function(){ cli_set_process_title(Config::SERVER_NAME . "-master"); });
        $http->on('WorkerStart', function ($serv, $worker_id){
            if($worker_id >= $serv->setting['worker_num']) {
                cli_set_process_title(Config::SERVER_NAME . "-task-" . $worker_id);
                //note task初始化
                Tasker::start($serv, $worker_id);
            } else {
                cli_set_process_title(Config::SERVER_NAME . "-worker-" . $worker_id);
                //note worker初始化
                Worker::start($serv, $worker_id);
            }
        });
        $http->on("request", "Application::onRequest"); //note 处理请求
        $http->on("task", "Tasker::onTask"); //note 处理task
        $http->on("finish", "Worker::onFinish"); //note task finish
        $http->start();
    }

    public static function getInstance($conf) {
        if (!self::$instance) {
            self::$instance = new HttpServer($conf);
        }
        return self::$instance;
    }
}
HttpServer::getInstance(Config::$swool);

配置文件

config_develop.php
<?php
class Config
{
    const ENV = "beta";

    const SERVER_NAME = "swool";
    const SWOOL_ADDR = "0.0.0.0";
    const SWOOL_PORT = 9505;

    public static $swool = array(
        "max_request" => 100000,
        "heartbeat_idle_time" => 600,
        "worker_num" => 4,
        "task_worker_num" => 1,
        "heartbeat_check_interval" => 60,
        "dispatch_mode" => 1,
        "log_file" => "/data/logs/php/swool.lvphp.com.log",
        "log_level" => 4,
        "daemonize" => 0
    );

    public static $redis = array(
        "cache.read"  => array(
            "name"    => "cache.read",
            "host"    => "127.0.0.1",
            "port"    => "14000",
            "pwd"     => "lvphp",
            "timeout" => 3,
            "prefix"  => "cache:",
        )
    );
}

swool.lvphp.com.conf
server {
    listen       80;
    server_name  swool.lvphp.com;
    root /data/htdocs/swool.lvphp.com/public/;
    index index.html;

    location / {
        if (!-e $request_filename) {
            proxy_pass http://127.0.0.1:9505;
        }
    }
    #access_log  /data/logs/nginx/swool.lvphp.com.access.log main;
}

类库

Application.php
<?php
defined('ROOT_PATH') || exit("NO ROOT_PATH!");
defined('LOG_PATH') || exit("NO LOG_PATH!");
defined('LOG_NAME') || exit("NO LOG_NAME!");
class Application{
    public static $starttime = null;
    public static $controller = null;
    public static $method = null;
    public static $argTypes = array("p"=>"post", "g"=>"get", "c"=>"cookie");

    public static $directories = array();
    public static $files = array();

    public static $inputs = null;
    public static $server = null;
    public function __construct() {}

    public static function load($loadData){
        if(isset($loadData["dir"])){ self::$directories = $loadData["dir"];}
        if(isset($loadData["file"])){ self::$files = $loadData["file"];}
        spl_autoload_register("Application::autoLoad");
    }
    public static function autoLoad($className){
        $dsClassName = str_replace("_", "/", $className);
        if(is_array(self::$directories)){
            foreach(self::$directories as $directorie){
                $filePath = rtrim($directorie, "/") . "/" . $dsClassName . ".php";
                if(is_file($filePath)){
                    require $filePath;
                    return true;
                }
            }
        }
        if(is_array(self::$files)){
            foreach(self::$files as $key => $file){
                if($key == $className){
                    require $file;
                    return true;
                }
            }
        }
        throw new Exception("No found class " . $className, 500);
    }

    public static function writeLog($type, $data, $addTime = true){
        $data = $addTime ? (date("Y-m-d H:i:s") . "\t" . $data) : $data;
        swoole_async_writefile(LOG_PATH . "/" . LOG_NAME . "_{$type}.log", $data . "\r\n", function(){}, FILE_APPEND);
    }

    public static function display($result){
        //note 请求响应日志
        $consume = (int)round((microtime(true) - Application::$starttime) * 1000, 1);
        $logInfo = self::$inputs;
        $logInfo['client_ip'] = self::getIP();
        $logInfo['logtime'] = date("Y-m-d H:i:s", intval(Application::$starttime));
        $logInfo['consume'] = $consume;
        $logInfo['errno'] = $result['errno'];
        $logInfo['errmsg'] = $result['errmsg'];
        self::writeLog('request', json_encode($logInfo), false);
        return json_encode($result);
    }
    public static function get($name, $hash = 'get', $default = null) {
        $hash = strtolower ( $hash );
        switch ($hash) {
            case 'get' : $input = self::$inputs['get'];break;
            case 'post' : $input = self::$inputs['post'];break;
            case 'cookie' : $input = self::$inputs['cookie'];break;
            case 'files' : $input = self::$inputs['files'];break;
            case 'server' : $input = self::$server;break;
            default : $input = isset(self::$inputs[$hash]) ? self::$inputs[$hash] : NULL;break;
        }
        return (isset ($input[$name]) ? $input[$name] : $default);
    }
    public static function set($name, $hash = 'get', $value = ""){
        self::$inputs[strtolower($hash)][$name] = $value;
    }
    public static function getIP(){
        $ip = false;
        if (getenv("REMOTE_ADDR") && strcasecmp(getenv("REMOTE_ADDR"), "unknown")){
            $ip = getenv("REMOTE_ADDR");
        }else if (isset(self::$server['remote_addr']) && self::$server['remote_addr'] && strcasecmp(self::$server['remote_addr'], "unknown")){
            $ip = self::$server['remote_addr'];
        }
        return $ip;
    }

    public static function Route(){
        $method = self::get('method');
        if($method){
            $info = explode(".", $method);
            if(count($info) == 2){
                $controllerName = ucfirst($info[0]) . 'Controller';
                $actionName = strtolower($info[1]) . 'Action';
                //note 实例化控制器
                self::$controller = new $controllerName();
                if(method_exists(self::$controller, $actionName)){
                    $className = get_class(self::$controller);
                    $class = new ReflectionClass($className);
                    self::$method = $class->getMethod($actionName);
                }else{
                    throw new Exception("No method exists", 500);
                }
            }else{
                throw new Exception("Error method param", 401);
            }
        }else{
            throw new Exception("No method param", 401);
        }
    }
    public static function prepareParams(){
        $args = array();
        $params = self::$method->getParameters();
        if($params){
            foreach($params as $p){
                $input = false;
                $pargName = $p->getName();
                $tmp = explode("_", $pargName);
                if(isset(self::$argTypes[$tmp[0]])){
                    $pargType = self::argTypes[$tmp[0]];
                    $pargName = substr($pargName, 2);
                }else{
                    $pargType = "get";
                }
                $input = isset( self::$inputs[$pargType][$pargName] ) ? self::$inputs[$pargType][$pargName] : false;
                $default = $p->isDefaultValueAvailable();
                if(0 != $input && (false === $input || empty($input)) && false === $default){
                    throw new Exception("No {$pargName} {$pargType} param", 401);
                }
                $args[$pargName] = false !== $input ? $input : $p->getDefaultValue();
            }
        }
        return $args;
    }

    public static function Run() {
        self::Route();
        return self::$method->invokeArgs(self::$controller, self::prepareParams());
    }

    public static function onRequest($req, $res){
        $result = array("errno"=>0, "errmsg"=>"", "data"=>"");
        self::$starttime = microtime(true);
        self::$server = $req->server;
        self::$inputs = array(
            "get" => $req->get,
            "post" => $req->post,
            "cookie" => $req->cookie,
            "files" => $req->files,
            "input" => empty($req->post) ? $req->rawContent() : null,
        );
        try{
            //note 程序运行
            $result['data'] = self::run();
        }catch(Exception $e){
            $errno = $e->getCode();
            empty($errno) && $errno = 500;
            $result['errno'] = $errno;
            $result['errmsg'] = $e->getMessage();
        }
        //note 输出结果
        $res->header("Content-Type", "application/json; charset=UTF-8");
        $res->header("Cache-Control", "no-store");
        $res->end(self::display($result));
    }
}

RedisDB.php
<?php
class RedisDB{
    public static $pool = null;
    public static function getInstance($conf){
        if($conf){
            $name = $conf['name'];
            if(!isset(self::$pool[$name])){
                try{
                    self::$pool[$name] = new Redis();
                    $result = self::$pool[$name]->pconnect($conf['host'], $conf['port'], $conf['timeout']);
                    if(!$result){
                        throw new Exception("Redis {$name} connect error", 500);
                    }
                    if(isset($conf['pwd'])){
                        self::$pool[$name]->auth($conf["pwd"]);
                    }
                    if(isset($conf['prefix'])){
                        self::$pool[$name]->setOption(Redis::OPT_PREFIX, $conf['prefix']);
                    }
                }catch(Exception $e){
                    throw new Exception(ErrorMessage::$errmsg[ErrorMessage::INVALID_REQUEST], ErrorMessage::INVALID_REQUEST);
                }
            }
            return self::$pool[$name];
        }else{
            throw new Exception("No Redis config", 500);
        }
    }
}

进程

Tasker.php
<?php

class Tasker{
    public static $serv = null;
    public static $tasker_id = null;

    public static $redisClient = null;

    public static function start($serv, $tasker_id){
        self::$serv = $serv;
        self::$tasker_id = $tasker_id;
        self::$redisClient = RedisDB::getInstance(Config::$redis["cache.read"]);
    }

    public static function onTask($serv, $task_id, $src_worker_id, $data){
        self::$redisClient->incr($data);
        self::$serv->finish("finish");
    }
}

Worker.php
<?php

class Worker{
    public static $serv = null;
    public static $worker_id = null;

    public static $redisClient = null;
    public static $rand;

    public static function start($serv, $worker_id){
        self::$serv = $serv;
        self::$worker_id = $worker_id;
        self::$redisClient = RedisDB::getInstance(Config::$redis["cache.read"]);
        self::$rand = rand(0, 999999);
    }

    public static function onFinish($serv, $task_id, $data){

    }
}

控制器

IndexController.php
<?php

class IndexController{

    public function indexAction($name = "hehe", $type){
        if($name =="aaa"){
            sleep(3);
        }
        Worker::$serv->taskwait("wbtest", $timeout = 0.5);
        $data = Worker::$redisClient->incr("wbtest");
        return Application::get("name") . "-" . Worker::$rand . "-" . $data;
    }

}

说明

以上代码只是一个demo，做了一下简单的路由，可以自己根据实际情况使用，task worker根据自己的需要确定是否使用。