Swoole http server + yaf, swoole socket server + protobuf 等小結

擁抱swoole, 擁抱更好的php

Swoole 是什麼？javascript

Yaf 是什麼？php

接觸swoole已經4年多了，一直沒有好好靜下心來學習。一直在作web端的應用，對網絡協議和常駐內存型服務器一竅不通。一不留神swoole已經從小衆擴展變成了流行框架，再不學習就完了css

swoole + yaf

###swoole server 的角色仍是先用swoole來作一個http server。常見的php web應用，一般是apache+fast-cgi 或者 nginx + php-fpm。這裏以php-fpm爲例，咱們配置nginx.conf的時候都要配置一個html

location ~*\.php$ {
        root            /usr/share/nginx/html;
        fastcgi_index   index.php;
        fastcgi_pass    127.0.0.1:9000;
        include         fastcgi_params;
        ...
}

主要是這句 fastcgi_pass 127.0.0.1:9000;。就是說nginx 匹配到請求的uri是php後綴的時候，就把http request 轉交給127.0.0.1:9000處理了。若是你查看或者修改過php-fpm的配置文件，就知道9000是php-fpm的默認端口。那麼到這裏咱們就清楚了，nginx把php文件交給php-fpm處理，php-fpm執行php腳本後返回http response給nginx。接下來就好理解swoole http server 的做用以及應該扮演的角色。swoole http server 本身接受http請求，處理靜態文件和php腳本，而後返回給客戶端。swoole server 的配置項中有一個 document_root 用來告訴swoole 從哪裏讀取靜態文件。固然，咱們仍然能夠用nginx來處理靜態文件，只把php腳本交給swoole處理，這裏須要修改nginx.conf，用nginx的代理功能 proxy_pass前端

location ~ .(gif|jpg|jpeg|png|bmp|swf|css|js)$ {  
    root     /data/www/swoole-server/public;  
} 
        
location / {
    proxy_http_version 1.1;
    proxy_set_header Connection "keep-alive";
    proxy_set_header X-Real-IP $remote_addr;
    proxy_pass http://127.0.0.1:9501;
}

以上說了這麼多，做爲一個php web開發人員，應該能夠大概理解日常寫的邏輯代碼，就是在swoole server 的 onRequest中。包括日常的PHP全局變量 _SERVER, _COOKIE _GET _POST 等等，都在swoole server 的回調函數的參數 Request 中。那麼咱們接下來在onRequest回調中，天然要解析 uri，而後作路由解析進入到具體的業務邏輯。最簡單的就是直接require uri的這個php腳本，也就是第一次接觸php的script模式。路由解析，加載控制器MVC渲染這些都是框架最擅長的事情，所以在onRequest中咱們引入框架，返回結果給swoole response對象。java

接入Yaf

Swoole 的worker子進程是實際的工做進程，在收到客戶端request的時候，swoole把request發送給worker，調用onRequest回調處理。若是咱們在onRequest中引入Yaf 建立yaf app對象，因爲onRequest是一個輪詢事件回調，worker會重複建立yaf app，yaf app實際上處於相同的上下文，所以會提示已經存在yaf application對象。並且，咱們並不須要在這裏重複讀取咱們的配置文件。咱們把yaf application 放在 onWorkerStart 中，一個worker 只產生一個yaf app對象，這個yaf對象輪詢處理request uri 。 Swoole Http Server onWorkerStart & onRequestnginx

public function onWorkerStart($serv, $work_id) {
    // var_dump(get_included_files()); // 打印worker啓動前已經加載的php文件
    cli_set_process_title('swoole_worker_'.$work_id); // 設置worker子進程名稱
    Yaf\Registry::set('swoole_serv', $serv);
    $this->app = new Yaf\Application( APPLICATION_PATH . "conf/application.ini");
    $this->app->bootstrap();
}

public function onRequest($request, $response) {
    // print_r($request->server);
    $uri = $request->server['request_uri'];
    printf("[%s]get %s\n", date('Y-m-d H:i:s'), $uri);
    if ($uri == '/favicon.ico') {
        $response->status(404);
        $response->end();
    } else {
        Yaf\Registry::set('swoole_req', $request);
        Yaf\Registry::set('swoole_res', $response);
        // yaf 會自動輸出腳本內容，所以這裏使用緩存區接受交給swoole response 對象返回
        ob_start();
        $this->app->getDispatcher()->dispatch(new Yaf\Request\Http($this->rewrite($uri))); // rewrite 中能夠應用本身的規則
        $data = ob_get_clean();
        $response->end(data);
    }
}

若是你用過yaf，接下來只須要寫一個標準的yaf框架應用就能夠了。yaf 框架的public文件夾再也不須要入口文件 index.php，nginx 中也再也不須要重寫uri規則，想一想爲啥git

Swoole WebSocket

理解了http server 以後，咱們再來建立一個websocket 服務器。websocket是web開發人員相對更熟悉的服務器，瀏覽器用javascript能夠寫一個現成的客戶端。swoole websocket服務器與http 服務器大同小異，只不過onRequest()方法變成了onMessage()，$response->end()變成了$server->push(); websocket是有狀態的長鏈接，http是無狀態的。無狀態意思是說http你只須要知道request是什麼，而後給他response，不論是誰，請求幾回request，都是同樣的response。而有狀態的意思是，對於每個請求，你須要分辨它是誰。所以對於相同的請求，可能會有不一樣的處理。websocket的每一個客戶端連接有惟一標識fd，有點相似於會話session id 的意思。與onRequest()方法相似，在onMessage()方法中，咱們須要對客戶端發送的數據進行路由解析，而後想客戶端返回結果。不過這裏再也不是http協議的url請求格式了，是咱們本身組裝的協議數據包，好比一個JSON結構，包括action,controller,module等等。咱們仍然能夠引入yaf框架，利用他的類庫自動加載Loader和路由Dispatcher機制，來處理客戶端請求，這裏再也不贅述。github

public function onMessage(\Swoole\Websocket\Server $serv, \Swoole\Websocket\Frame $frame) {
    $route = json_decode($frame->data);
    if ($route->module) {
        try {
            ob_start();
            $this->app->getDispatcher()->dispatch(new Yaf\Request\Simple('cli', $route->module, $route->controller, $route->action, $route->params));
            $response = ob_get_clean();
        } catch (Exception $e) {
            // handle exception
        }
        $serv->push($frame->fd, $response);
    } else {
        printf("[%s] unknow message: %s\n", date('Y-m-d H:i:s'), $frame->data);
    }
}

PHP 使用 Protobuf 消息

上面咱們使用了一個 JSON 協議傳輸websocket的例子，而 Protobuf 是與JSON 相似的一種消息協議，除此以外，你們熟知的xml也是一種消息協議。ProtoBuf 是google開源的一種通訊協議，既然是google的，那麼別問，學就對了。web

什麼是ProtoBuf

相比JSON與XML，ProtoBuf的好處體如今

解析快。爲何比XML，JSON的字符串解析快呢，google大神們說快那就是快，別問。
節省包體大小。它把咱們的消息結構體轉爲二進制流進行傳輸，到了另外一端再經過相同的結構體定義解析還原。
自然的消息加密。傳輸過程當中是二進制，xml或者json還須要進一步加解密才能保密。與之同時帶來的缺點，就是可讀性差。你看着一堆二進制串，在消息解析出來以前徹底不知道發的是啥（我的認爲並非什麼缺點）。

php 處理protobuf

用php處理protobuf咱們須要用到兩個東西

protoc https://repo1.maven.org/maven2/com/google/protobuf/protoc/ protoc 是將proto結構體文件轉換成對應的php文件，每一個文件就是一個消息體類 /path/protoc --php_dir=/php-lib/ xxx.proto
proto-php 擴展（類庫）https://github.com/protocolbuffers/protobuf/tree/master/php protoc 只是負責將proto文件轉成php類，這些類的父類定義及使用須要php安裝protobuf擴展，或者在項目中直接引入php類庫（擴展和類庫的概念應該知道的吧。。）

咱們在解析protobuf二進制流以前，是須要先指定對應的消息結構體的，所以咱們不能只發送一個protobuf，至少應該再附帶一個消息ID。經過這個消息ID對應的結構體，咱們才能解析具體的protobuf消息。 php處理二進制數據須要用到pack()和unpack()。若是像我同樣沒接觸過的同窗，能夠臨時補補課，學習一下字節序什麼的

PHP中pack、unpack的詳細用法

假設咱們有一個int32位無符號消息ID，那麼每一個包體的結構就是 消息ID+protobuf。發送消息以前，咱們進行數據打包

public function pack($msg_id, $msg_body) {
    $proto_class = Proto::GetResponseMessageProto($msg_id); // 由消息ID獲取對應的proto結構體類名
    if (!$proto_class ) {
        $this->err = 'No msg id matched.';
        return FALSE;
    }
    try {
        $msg_obj = new $proto_class ();
        // 定義消息
        $msg_obj->mergeFromArray($msg_body);
        // 打包protobuf
        $buf_str= $msg_obj->serializeToString();
        // 拼接消息體
        $this->bufString = pack('N', $msg_id). $buf_str;;
        return TRUE;
    } catch (\Exception $e){
        $this->err = $e->getMessage();
        return FALSE;
    }
}

數據打包相對簡單些，數據解包會有一點曲折。也就是在這裏我感受PHP在處理二進制數據上有點侷限，也多是我沒有掌握更高效的方法。若是有的話，還望各位讀者不吝賜教。

public function unpack($msg) {
    $data = unpack('Nmsg_id/a*msg_body', $msg);
    $msg_id = $data['msg_id'];
    // 暫時把protobuf解析成字符串
    $buf_str = $data['msg_body'];
    $proto_class = Proto::GetRequestMessageProto($msg_id);
    if (!$proto_class) {
        $this->err = 'No msg id matched.';
        return FALSE;
        // handle error.
    }
    try {
        $msg_obj = new $proto_class();
        // 上面已經把probuf解析成了字符串，所以這裏須要再轉化爲二進制
        $msg_obj->mergeFromString(pack('a*', $buf_str));
        print_r($msg_obj->serializeToJsonString()); // protobuf 類的讀取接口比較少，建議去看看源碼
    } catch (\Exception $e) {
        $this->err = $e->getMessage();
        return FALSE;
        // handle invalid msg
//            throw new MessageParseException('Invalid message');
    }
    $this->msg_obj = $msg_obj->serializeToJsonString(); // 消息體
    $this->msg_id = $msg_id; // 消息ID
    return TRUE;
}

接收消息的處理

// onMessage
public function onMessage(swoole_websocket_server $serv, swoole_websocket_frame $frame) {
    $msg = new \Message\Message();
    if ($msg->unpack($frame->data)) {
        printf("[%s] receive data: %d %s\n", date('Y-m-d H:i:s'), $msg->msg_id, $msg->msg_obj);
        // dispatcher
        list($module, $controller, $action) = $this->dispatch($msg->msg_id); // 本身的消息路由，就是某一個消息ID交給哪一個控制器進行處理
        try {
            ob_start();
            $this->app->getDispatcher()->dispatch(new Yaf\Request\Simple('cli', $module, $controller, $action, json_decode($msg->msg_obj, TRUE)));
            $response = ob_get_clean();
            $code = 0;
        } catch (Exception $e) {
            $response = json_encode(['err' => $e->getMessage()]);
            $code = -1;
        }
        print_r($response);
        if (!$msg->pack($msg->msg_id, $response)) {
            print_r('msg pack err:'. $msg->err);
        } else {
            $serv->push($frame->fd, $msg->bufString, WEBSOCKET_OPCODE_BINARY); // websocket 發送二進制
        }
    } else {
        printf("[%s] unpack err: %s\n", date('Y-m-d H:i:s'), $frame->data);
        print_r('msg unpack err:'. $msg->err);
    }
}

附前端javascript的示例

javascript處理相對來講還更簡單，用到的是 ArrayBuffer

var protoRoot = null;
protobuf.load('/data/game.proto', function(err, root) {
    if (err)
        throw err;
    protoRoot = root;
});
function writeBuf(msgid, buf) {
    // buf 是protobuf消息的二進制結果
    var length = buf.length;
    var buffer = new ArrayBuffer(buf.length + 4); // 消息ID佔4位
    var dv = new DataView(buffer);
    dv.setUint32(0, msgid, false); // 大端字節序
    for (let i=0;i<buf.length;i++) {
        dv.setInt8(4+i, buf[i]); // 逐字節寫入buffer
    }
    console.log(buffer);
    return buffer;
}
function readBuf(buf) {
    var dv = new DataView(buf);
    var msgid = dv.getUint32(0, false);
    var buf = new Uint8Array(buf, 4); // 截取消息ID後面的字節，交給protobuf解析
    return [msgid, buf];
}
function Request_Message(msg, req, callback) {
    // 將客戶端請求的消息msg轉成protobuf 
    var RequestMessage = protoRoot.lookupType("dapianzi."+req); // 這裏須要加上命名空間
    var errMsg = RequestMessage.verify(msg);
    if (errMsg)
        throw Error(errMsg);
    var message = RequestMessage.fromObject(msg);
    var buffer = RequestMessage.encode(message).finish();
    callback(buffer); // 下一步調用writeBuf 產生消息包，發送給服務器
}
function Response_Message(buf, res, callback) {
    // buf 是readBuf()中返回的二進制串，這裏交給protobuf解析成消息體
    var ResponseMessage = protoRoot.lookupType("dapianzi."+res);
    var message = ResponseMessage.decode(buf);
    var object = ResponseMessage.toObject(message, {
        longs: String,
        enums: String,
        bytes: String,
    });
    callback(object); // 進行客戶端邏輯
}

後記

在websocket服務器中使用yaf仍是以爲比較牽強，畢竟yaf是一個web框架，使用它僅僅是能夠比較方便的使用lib自動加載，以及路由映射。所以，仍是得本身想辦法寫一個簡單的框架，實現消息路由，類庫加載，事件註冊，和全局對象的容器管理。