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

时间  2021-02-17
标签 javascript php css html 前端 java nginx git github web 栏目 HTTP/TCP 繁體版
原文   原文链接

拥抱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咱们须要用到两个东西

咱们在解析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自动加载，以及路由映射。所以，仍是得本身想办法写一个简单的框架，实现消息路由，类库加载，事件注册，和全局对象的容器管理。

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程