• EasySwoole Tcp服务
    • 创建tcp服务
      • 粘包问题
      • 粘包解决
    • 实现粘包处理
      • 服务端:
      • 客户端:
    • tcp控制器实现
      • 协议规则与解析
      • 实现控制器 Index.php
    • 开启子服务
    • 测试客户端
    • HTTP往TCP推送

    EasySwoole Tcp服务

    tcp 服务以及tcp客户端 demo:https://github.com/easy-swoole/demo/tree/3.x-subtcp

    创建tcp服务

    通过EasySwooleEvent.php文件的mainServerCreate 事件,进行添加子服务监听,例如:

    1. <?php
    2. public static function mainServerCreate(EventRegister $register)
    3. {
    4.     $server = ServerManager::getInstance()->getSwooleServer();
    6.     ################# tcp 服务器1 没有处理粘包 #####################
    7.     $subPort1 = $server->addlistener('0.0.0.0', 9502, SWOOLE_TCP);
    8.     $subPort1->set(
    9.         [
    10.             'open_length_check' => false,//不验证数据包
    11.         ]
    12.     );
    13.     $subPort1->on('connect', function (\swoole_server $server, int $fd, int $reactor_id) {
    14.         echo "tcp服务1  fd:{$fd} 已连接\n";
    15.         $str = '恭喜你连接成功服务器1';
    16.         $server->send($fd, $str);
    17.     });
    18.     $subPort1->on('close', function (\swoole_server $server, int $fd, int $reactor_id) {
    19.         echo "tcp服务1  fd:{$fd} 已关闭\n";
    20.     });
    21.     $subPort1->on('receive', function (\swoole_server $server, int $fd, int $reactor_id, string $data) {
    22.         echo "tcp服务1  fd:{$fd} 发送消息:{$data}\n";
    23.     });
    24. }

    粘包问题

    由于tcp的特性,可能会出现数据粘包情况,例如

    • A连接Server
    • A发送 hello
    • A又发送了一条 hello
    • Server可能会一次性收到一条”hellohello”的数据
    • Server也可能收到”he” ,”llohello”类似这样的中断数据

    粘包解决

    • 通过标识EOF,例如http协议,通过\r\n\r\n 的方式去表示该数据已经完结,我们可以自定义一个协议,例如当接收到 “结尾666” 字符串时,代表该字符串已经结束,如果没有获取到,则存入缓冲区,等待结尾字符串,或者如果获取到多条,则通过该字符串剪切出其他数据
    • 定义消息头,通过特定长度的消息头进行获取,例如我们定义一个协议,前面10位字符串都代表着之后数据主体的长度,那么我们传输数据时,只需要000000000512346(前10位为协议头,表示了这条数据的大小,后面的为数据),每次我们读取只先读取10位,获取到消息长度,再读取消息长度那么多的数据,这样就可以保证数据的完整性了.(但是为了不被混淆,协议头也得像EOF一样标识)
    • 通过pack二进制处理,相当于于方法2,将数据通过二进制封装拼接进消息中,通过验证二进制数据去读取信息,sw采用的就是这种方式

    可查看swoole官方文档:https://wiki.swoole.com/wiki/page/287.html

    实现粘包处理

    服务端:

    1. <?php
    2.  ################# tcp 服务器2 没有处理粘包 #####################
    3.         $subPort2 = $server->addlistener('0.0.0.0', 9503, SWOOLE_TCP);
    4.         $subPort2->set(
    5.             [
    6.                 'open_length_check'     => true,
    7.                 'package_max_length'    => 81920,
    8.                 'package_length_type'   => 'N',
    9.                 'package_length_offset' => 0,
    10.                 'package_body_offset'   => 4,
    11.             ]
    12.         );
    13.         $subPort2->on('connect', function (\swoole_server $server, int $fd, int $reactor_id) {
    14.             echo "tcp服务2  fd:{$fd} 已连接\n";
    15.             $str = '恭喜你连接成功服务器2';
    16.             $server->send($fd, pack('N', strlen($str)) . $str);
    17.         });
    18.         $subPort2->on('close', function (\swoole_server $server, int $fd, int $reactor_id) {
    19.             echo "tcp服务2  fd:{$fd} 已关闭\n";
    20.         });
    21.         $subPort2->on('receive', function (\swoole_server $server, int $fd, int $reactor_id, string $data) {
    22.             echo "tcp服务2  fd:{$fd} 发送原始消息:{$data}\n";
    23.             echo "tcp服务2  fd:{$fd} 发送消息:" . substr($data, '4') . "\n";
    24.         });

    客户端:

    1. <?php
    2. /**
    3.  * Created by PhpStorm.
    4.  * User: Tioncico
    5.  * Date: 2019/3/6 0006
    6.  * Time: 16:22
    7.  */
    8. include "../vendor/autoload.php";
    9. define('EASYSWOOLE_ROOT', realpath(dirname(getcwd())));
    10. \EasySwoole\EasySwoole\Core::getInstance()->initialize();
    11. /**
    12.  * tcp 客户端2,验证数据包,并处理粘包
    13.  */
    14. go(function () {
    15.     $client = new \Swoole\Client(SWOOLE_SOCK_TCP);
    16.     $client->set(
    17.         [
    18.             'open_length_check'     => true,
    19.             'package_max_length'    => 81920,
    20.             'package_length_type'   => 'N',
    21.             'package_length_offset' => 0,
    22.             'package_body_offset'   => 4,
    23.         ]
    24.     );
    25.     if (!$client->connect('127.0.0.1', 9503, 0.5)) {
    26.         exit("connect failed. Error: {$client->errCode}\n");
    27.     }
    28.     $str = 'hello world';
    29.     $client->send(encode($str));
    30.     $data = $client->recv();//服务器已经做了pack处理
    31.     var_dump($data);//未处理数据,前面有4 (因为pack 类型为N)个字节的pack
    32.     $data = decode($data);//需要自己剪切解析数据
    33.     var_dump($data);
    34. //    $client->close();
    35. });
    37. /**
    38.  * 数据包 pack处理
    39.  * encode
    40.  * @param $str
    41.  * @return string
    42.  * @author Tioncico
    43.  * Time: 9:50
    44.  */
    45. function encode($str)
    46. {
    47.     return pack('N', strlen($str)) . $str;
    48. }
    50. function decode($str)
    51. {
    52.     $data = substr($str, '4');
    53.     return $data;
    54. }

    tcp控制器实现

    协议规则与解析

    在本文档中,传输json数据 使用pack N进行二进制处理,json数据有3个参数,例如:

    1. {"controller":"Index","action":"index","param":{"name":"\u4ed9\u58eb\u53ef"}}

    实现解析器Parser.php

    1. <?php
    2. /**
    3.  * Created by PhpStorm.
    4.  * User: Tioncico
    5.  * Date: 2018/10/17 0017
    6.  * Time: 9:10
    7.  */
    8. namespace App\TcpController;
    9. use EasySwoole\Socket\Bean\Caller;
    10. use EasySwoole\Socket\Bean\Response;
    11. use EasySwoole\Socket\AbstractInterface\ParserInterface;
    12. use EasySwoole\Utility\CommandLine;
    13. class Parser implements ParserInterface
    14. {
    15.     public function decode($raw, $client): ?Caller
    16.     {
    17.         $data = substr($raw, '4');
    18.         //为了方便,我们将json字符串作为协议标准
    19.         $data = json_decode($data, true);
    20.         $bean = new Caller();
    21.         $controller = !empty($data['controller']) ? $data['controller'] : 'Index';
    22.         $action = !empty($data['action']) ? $data['action'] : 'index';
    23.         $param = !empty($data['param']) ? $data['param'] : [];
    24.         $controller = "App\\TcpController\\{$controller}";
    25.         $bean->setControllerClass($controller);
    26.         $bean->setAction($action);
    27.         $bean->setArgs($param);
    28.         return $bean;
    29.     }
    30.     /**
    31.      * 只处理pack,json交给控制器
    32.      * encode
    33.      * @param Response $response
    34.      * @param          $client
    35.      * @return string|null
    36.      * @author Tioncico
    37.      * Time: 10:33
    38.      */
    39.     public function encode(Response $response, $client): ?string
    40.     {
    41.         return pack('N', strlen($response->getMessage())) . $response->getMessage();
    42.     }
    43. }

    实现控制器Index.php

    1. <?php
    2. /**
    3.  * Created by PhpStorm.
    4.  * User: Tioncico
    5.  * Date: 2018/10/17 0017
    6.  * Time: 9:15
    7.  */
    8. namespace App\TcpController;
    9. use App\Rpc\RpcServer;
    10. use EasySwoole\EasySwoole\ServerManager;
    11. use EasySwoole\EasySwoole\Swoole\Task\TaskManager;
    12. use EasySwoole\Socket\AbstractInterface\Controller;
    13. use http\Env\Response;
    14. class Index extends Controller{
    15.     function actionNotFound(?string $actionName)
    16.     {
    17.         $this->response()->setMessage("{$actionName} not found \n");
    18.     }
    19.     public function index(){
    20.         $this->response()->setMessage(time());
    21.     }
    22.     public function args()
    23.     {
    24.         $this->response()->setMessage('your args is:'.json_encode($this->caller()->getArgs()).PHP_EOL);
    25.     }
    26.     public function delay()
    27.     {
    28.         $client = $this->caller()->getClient();
    29.         TaskManager::async(function ()use($client){
    30.             sleep(1);
    31.             ServerManager::getInstance()->getSwooleServer()->send($client->getFd(),'this is delay message at '.time());
    32.         });
    33.     }
    34.     public function close()
    35.     {
    36.         $this->response()->setMessage('you are goging to close');
    37.         $client = $this->caller()->getClient();
    38.         TaskManager::async(function ()use($client){
    39.             sleep(2);
    40.             ServerManager::getInstance()->getSwooleServer()->send($client->getFd(),'this is delay message at '.time());
    41.         });
    42.     }
    43.     public function who()
    44.     {
    45.         $this->response()->setMessage('you fd is '.$this->caller()->getClient()->getFd());
    46.     }
    47. }

    开启子服务

    在EasySwooleEvent中注册。

    1. <?php
    2. public static function mainServerCreate(EventRegister $register)
    3. {
    4.        ############# tcp 服务器3 tcp控制器实现+处理粘包############
    6.               $subPort3 = $server->addListener(Config::getInstance()->getConf('MAIN_SERVER.LISTEN_ADDRESS'), 9504, SWOOLE_TCP);
    8.               $socketConfig = new \EasySwoole\Socket\Config();
    9.               $socketConfig->setType($socketConfig::TCP);
    10.               $socketConfig->setParser(new \App\TcpController\Parser());
    11.               //设置解析异常时的回调,默认将抛出异常到服务器
    12.               $socketConfig->setOnExceptionHandler(function ($server, $throwable, $raw, $client, $response) {
    13.                   echo  "tcp服务3  fd:{$client->getFd()} 发送数据异常 \n";
    14.                   $server->close($client->getFd());
    15.               });
    16.               $dispatch = new \EasySwoole\Socket\Dispatcher($socketConfig);
    18.               $subPort3->on('receive', function (\swoole_server $server, int $fd, int $reactor_id, string $data) use ($dispatch) {
    19.                   echo "tcp服务3  fd:{$fd} 发送消息:{$data}\n";
    20.                   $dispatch->dispatch($server, $data, $fd, $reactor_id);
    21.               });
    22.               $subPort3->set(
    23.                   [
    24.                       'open_length_check'     => true,
    25.                       'package_max_length'    => 81920,
    26.                       'package_length_type'   => 'N',
    27.                       'package_length_offset' => 0,
    28.                       'package_body_offset'   => 4,
    29.                   ]
    30.               );
    31.               $subPort3->on('connect', function (\swoole_server $server, int $fd, int $reactor_id) {
    32.                   echo "tcp服务3  fd:{$fd} 已连接\n";
    33.               });
    34.               $subPort3->on('close', function (\swoole_server $server, int $fd, int $reactor_id) {
    35.                   echo "tcp服务3  fd:{$fd} 已关闭\n";
    36.               });
    38. }

    测试客户端

    1. <?php
    2. /**
    3.  * Created by PhpStorm.
    4.  * User: Tioncico
    5.  * Date: 2019/3/6 0006
    6.  * Time: 16:22
    7.  */
    8. include "../vendor/autoload.php";
    9. define('EASYSWOOLE_ROOT', realpath(dirname(getcwd())));
    10. \EasySwoole\EasySwoole\Core::getInstance()->initialize();
    11. /**
    12.  * tcp 客户端3,验证数据包处理粘包 以及转发到控制器写法
    13.  */
    14. go(function () {
    15.     $client = new \Swoole\Client(SWOOLE_SOCK_TCP);
    16.     $client->set(
    17.         [
    18.             'open_length_check'     => true,
    19.             'package_max_length'    => 81920,
    20.             'package_length_type'   => 'N',
    21.             'package_length_offset' => 0,
    22.             'package_body_offset'   => 4,
    23.         ]
    24.     );
    25.     if (!$client->connect('127.0.0.1', 9504, 0.5)) {
    26.         exit("connect failed. Error: {$client->errCode}\n");
    27.     }
    29.     $data = [
    30.         'controller' => 'Index',
    31.         'action'     => 'index',
    32.         'param'      => [
    33.             'name' => '仙士可'
    34.         ],
    35.     ];
    36.     $str = json_encode($data);
    37.     var_dump($str);
    38.     $client->send(encode($str));
    39.     $data = $client->recv();//服务器已经做了pack处理
    40.     $data = decode($data);//需要自己剪切解析数据
    41.     echo "服务端回复: $data \n";
    43.     $data = [
    44.         'controller' => 'Index',
    45.         'action'     => 'args',
    46.         'param'      => [
    47.             'name' => '仙士可'
    48.         ],
    49.     ];
    50.     $str = json_encode($data);
    51.     $client->send(encode($str));
    52.     $data = $client->recv();//服务器已经做了pack处理
    53.     $data = decode($data);//需要自己剪切解析数据
    54.     echo "服务端回复: $data \n";
    59. //    $client->close();
    60. });
    62. /**
    63.  * 数据包 pack处理
    64.  * encode
    65.  * @param $str
    66.  * @return string
    67.  * @author Tioncico
    68.  * Time: 9:50
    69.  */
    70. function encode($str)
    71. {
    72.     return pack('N', strlen($str)) . $str;
    73. }
    75. function decode($str)
    76. {
    77.     $data = substr($str, '4');
    78.     return $data;
    79. }

    HTTP往TCP推送

    HTTP控制器

    1. <?php
    2. /**
    3.  * Created by PhpStorm.
    4.  * User: Apple
    5.  * Date: 2018/11/1 0001
    6.  * Time: 11:10
    7.  */
    9. namespace App\HttpController;
    13. use EasySwoole\EasySwoole\ServerManager;
    14. use EasySwoole\Http\AbstractInterface\Controller;
    16. class Index extends Controller
    17. {
    18.     function index()
    19.     {
    20.         // TODO: Implement index() method.
    21.     }
    23.     function push(){
    24.         $fd = intval($this->request()->getRequestParam('fd'));
    25.         $info = ServerManager::getInstance()->getSwooleServer()->connection_info($fd);
    26.         if(is_array($info)){
    27.             ServerManager::getInstance()->getSwooleServer()->send($fd,'push in http at '.time());
    28.         }else{
    29.             $this->response()->write("fd {$fd} not exist");
    30.         }
    31.     }
    32. }

    实际生产中，一般是用户TCP连接上来后，做验证，然后以userName=>fd的格式，存在redis中，需要http，或者是其他地方，比如定时器往某个连接推送的时候，就是以userName去redis中取得对应的fd，再send。注意，通过addServer形式创建的子服务器，以再完全注册自己的网络事件，你可以注册onclose事件，然后在连接断开的时候，删除userName=>fd对应。