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

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

拥抱swoole, 拥抱更好的php

Swoole 是什么？

Yaf 是什么？

接触swoole已经4年多了，一直没有好好静下心来学习。一直在做web端的应用，对网络协议和常驻内存型服务器一窍不通。一不留神swoole已经从小众扩展变成了流行框架，再不学习就完了

swoole + yaf

swoole server 的角色

还是先用swoole来做一个http server。

常见的php web应用，通常是apache+fast-cgi 或者 nginx + php-fpm。这里以php-fpm为例，我们配置nginx.conf的时候都要配置一个

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对象。

接入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 & onRequest

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规则，想想为啥

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机制，来处理客户端请求，这里不再赘述。

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的，那么别问，学就对了。

什么是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); } }