RPC服务

EasySwoole 提供开放式的RPC服务。RPC服务分为三级模式：服务=>服务组=>行为。每个服务可以单独现在Openssl加密。
支持超时、成功、失败回调（即最基础的熔断保护和服务降级）

示例代码

服务端

服务A

  1. namespace App\RpcController\A;
  4. use EasySwoole\Core\Component\Rpc\AbstractInterface\AbstractRpcService;
  6. class G extends AbstractRpcService
  7. {
  8.     function index()
  9.     {
  10.         // TODO: Implement index() method.
  11.         $this->response()->setArgs([12,3]);
  12.     }
  13. }

服务A中存在G服务组，G服务组中实现了Index行为。

服务B

  1. namespace App\RpcController\B;
  4. use EasySwoole\Core\Component\Rpc\AbstractInterface\AbstractRpcService;
  6. class Index extends AbstractRpcService
  7. {
  9.     function index()
  10.     {
  11.         // TODO: Implement index() method.
  12.         var_dump('hit');
  13.         $this->response()->setResult('this is b index');
  14.     }
  15. }

服务绑定

  1. use EasySwoole\Core\Component\Rpc\Server;
  2. public static function mainServerCreate(ServerManager $server,EventRegister $register): void
  3. {
  4.     // TODO: Implement mainServerCreate() method.
  5.     Server::getInstance()->addService('A',9502)
  6.                 ->addService('B',9503,'password123')
  7.                 ->attach();
  8. }

客户端

客户端测试代码

  1. require_once 'vendor/autoload.php';
  3. \EasySwoole\Core\Core::getInstance()->initialize();
  4. //注册服务，让RPC服务管理中心知道当前系统中存在哪些服务
  6. $ServiceManager = \EasySwoole\Core\Component\Rpc\Server::getInstance();
  7. $ServiceManager->updateServiceNode(new \EasySwoole\Core\Component\Rpc\Common\ServiceNode(
  8.     [
  9.         'serviceName'=>'A',
  10.         'port'=>9502
  11.     ]
  12. ));
  14. $ServiceManager->updateServiceNode(new \EasySwoole\Core\Component\Rpc\Common\ServiceNode(
  15.     [
  16.         'serviceName'=>'B',
  17.         'port'=>9503,
  18.         'encryptToken'=>'password123'
  19.     ]
  20. ));
  23. //创建RPC客户端
  24. $client = new \EasySwoole\Core\Component\Rpc\Client();
  26. //调用A服务中G服务组的index行为
  27. $client->addCall('A','g','index')->setFailCall(function(\EasySwoole\Core\Component\Rpc\Client\ServiceResponse $response){
  28.     var_dump('11fail',$response);
  29. })->setSuccessCall(function (\EasySwoole\Core\Component\Rpc\Client\ServiceResponse $response){
  30.     var_dump('11success',$response);
  31. });
  33. //调用A服务中G服务组的c行为
  34. $client->addCall('A','g','c')->setFailCall(function(\EasySwoole\Core\Component\Rpc\Client\ServiceResponse $response){
  35.     var_dump('22fail',$response);
  36. })->setSuccessCall(function (\EasySwoole\Core\Component\Rpc\Client\ServiceResponse $response){
  37.     var_dump('22success',$response);
  38. });
  39. //调用A服务中c服务组的c行为
  40. $client->addCall('A','c','c')->setFailCall(function(\EasySwoole\Core\Component\Rpc\Client\ServiceResponse $response){
  41.     var_dump('33fail',$response);
  42. })->setSuccessCall(function (\EasySwoole\Core\Component\Rpc\Client\ServiceResponse $response){
  43.     var_dump('33success',$response);
  44. });
  45. //调用c服务中c服务组的c行为
  46. $client->addCall('c','c','c')->setFailCall(function(\EasySwoole\Core\Component\Rpc\Client\ServiceResponse $response){
  47.     var_dump('44fail',$response);
  48. })->setSuccessCall(function (\EasySwoole\Core\Component\Rpc\Client\ServiceResponse $response){
  49.     var_dump('44success',$response);
  50. });
  52. //调用B服务中c服务组的index行为
  53. $client->addCall('B','c','index')->setFailCall(function(\EasySwoole\Core\Component\Rpc\Client\ServiceResponse $response){
  54.     var_dump('55fail',$response);
  55. })->setSuccessCall(function (\EasySwoole\Core\Component\Rpc\Client\ServiceResponse $response){
  56.     var_dump('55success',$response);
  57. });
  59. //执行调用
  60. $client->call();

在没有集群模式时，可以在EasySwooleEvent的主服务启动事件中，注册好存在的服务信息，以后客户端可以直接调用服务，不需要继续再做服务发现注册。
再集群模式时，有服务自动发现。

原理讲解

服务端

服务端实现关键代码

  1. namespace EasySwoole\Core\Component\Rpc;
  4. use EasySwoole\Core\AbstractInterface\Singleton;
  5. use EasySwoole\Core\Component\Openssl;
  6. use EasySwoole\Core\Component\Rpc\Common\Parser;
  7. use EasySwoole\Core\Component\Rpc\Common\ServiceResponse;
  8. use EasySwoole\Core\Component\Rpc\Common\Status;
  9. use EasySwoole\Core\Component\Rpc\Server\ServiceManager;
  10. use EasySwoole\Core\Component\Rpc\Server\ServiceNode;
  11. use EasySwoole\Core\Component\Trigger;
  12. use EasySwoole\Core\Socket\Client\Tcp;
  13. use EasySwoole\Core\Socket\Response;
  14. use EasySwoole\Core\Swoole\EventHelper;
  15. use EasySwoole\Core\Swoole\ServerManager;
  17. class Server
  18. {
  19.     use Singleton;
  21.     private $list = [];
  22.     private $controllerNameSpace = 'App\\RpcController\\';
  24.     private $protocolSetting = [
  25.         'open_length_check' => true,
  26.         'package_length_type'   => 'N',
  27.         'package_length_offset' => 0,
  28.         'package_body_offset'   => 4,
  29.         'package_max_length'    => 1024*64,
  30.         'heartbeat_idle_time' => 5,
  31.         'heartbeat_check_interval' => 30,
  32.     ];
  33.     //可以自定义分包协议，这部分功能的parser 暂未分离，提前预留
  34.     function setProtocolSetting(array $data)
  35.     {
  36.         $this->protocolSetting = $data;
  37.         return $this;
  38.     }
  39.     //自定义RPC控制器名称空间
  40.     function setControllerNameSpace(string $nameSpace):Server
  41.     {
  42.         $this->controllerNameSpace = $nameSpace;
  43.         return $this;
  44.     }
  45.     //添加一个服务
  46.     function addService(string $serviceName,int $port,$encryptToken = null,string $address = '0.0.0.0')
  47.     {
  48.         //一个EasySwoole服务上不允许同名服务
  49.         $this->list[$serviceName] = [
  50.             'serviceName'=>$serviceName,
  51.             'port'=>$port,
  52.             'encryptToken'=>$encryptToken,
  53.             'address'=>$address
  54.         ];
  55.         return $this;
  56.     }
  57.     //绑定到主服务
  58.     public function attach()
  59.     {
  60.         foreach ($this->list as $name => $item){
  61.             $node = new ServiceNode();
  62.             $node->setPort($item['port']);
  63.             $node->setServiceName($name);
  64.             $node->setEncryptToken($item['encryptToken']);
  65.             ServiceManager::getInstance()->addServiceNode($node);
  67.             $sub = ServerManager::getInstance()->addServer("RPC_SERVER_{$name}",$item['port'],SWOOLE_TCP,$item['address'],$this->protocolSetting);
  69.             $nameSpace = $this->controllerNameSpace.ucfirst($item['serviceName']);
  70.             EventHelper::register($sub,$sub::onReceive,function (\swoole_server $server, int $fd, int $reactor_id, string $data)use($item,$nameSpace){
  71.                 $response = new ServiceResponse();
  72.                 $client = new Tcp($fd,$reactor_id);
  73.                 //解包，获得原始完整字符串
  74.                 $data = Parser::unPack($data);
  75.                 $openssl = null;
  76.                 //若有加密配置，则对数据包解密
  77.                 if(!empty($item['encryptToken'])){
  78.                     $openssl = new Openssl($item['encryptToken']);
  79.                 }
  80.                 if($openssl){
  81.                     $data = $openssl->decrypt($data);
  82.                 }
  83.                 if($data !== false){
  84.                     //看看能否成功解析出命令
  85.                     $caller = Parser::decode($data,$client);
  86.                     if($caller){
  87.                         $response->arrayToBean($caller->toArray());
  88.                         $response->setArgs(null);
  89.                         $group = ucfirst($caller->getServiceGroup());
  90.                         //搜索有没有完整的服务=>服务组控制器
  91.                         $controller = "{$nameSpace}\\{$group}";
  92.                         if(!class_exists($controller)){
  93.                             $response->setStatus(Status::SERVICE_GROUP_NOT_FOUND);
  94.                             //若没有，则搜索有没有完整的服务Index控制器（默认Index服务组）
  95.                             $controller = "{$nameSpace}\\Index";
  96.                             if(!class_exists($controller)){
  97.                                 $controller = null;
  98.                             }else{
  99.                                 $response->setStatus(Status::OK);
  100.                             }
  101.                         }
  102.                         if($controller){
  103.                             try{
  104.                                 (new $controller($client,$caller,$response));
  105.                             }catch (\Throwable $throwable){
  106.                                 Trigger::throwable($throwable);
  107.                                 $response->setStatus(Status::SERVICE_ERROR);
  108.                             }
  109.                         }else{
  110.                             $response->setStatus(Status::SERVICE_NOT_FOUND);
  111.                         }
  112.                     }else{
  113.                         $response->setStatus(Status::PACKAGE_DECODE_ERROR);
  114.                     }
  115.                 }else{
  116.                     $response->setStatus(Status::PACKAGE_ENCRYPT_DECODED_ERROR);
  117.                 }
  118.                 //进行json打包，并放回给客户端
  119.                 $response = json_encode($response->toArray(),JSON_UNESCAPED_SLASHES|JSON_UNESCAPED_UNICODE);
  120.                 if($openssl){
  121.                     $response =  $openssl->encrypt($response);
  122.                 }
  123.                 Response::response($client,Parser::pack($response));
  124.             });
  125.         }
  126.     }
  127. }

客户端

客户端基于swoole client + socket select实现的伪异步客户端。

跨平台调用

json请求结构体

  1. namespace EasySwoole\Core\Component\Rpc\Common;
  4. use EasySwoole\Core\Component\Spl\SplBean;
  6. class ServiceCaller extends SplBean
  7. {
  8.     protected $serviceName;
  9.     protected $serviceGroup;
  10.     protected $serviceAction;
  11.     protected $args = null;
  13.     /**
  14.      * @return mixed
  15.      */
  16.     public function getServiceName()
  17.     {
  18.         return $this->serviceName;
  19.     }
  21.     /**
  22.      * @param mixed $serviceName
  23.      */
  24.     public function setServiceName($serviceName): void
  25.     {
  26.         $this->serviceName = $serviceName;
  27.     }
  29.     /**
  30.      * @return mixed
  31.      */
  32.     public function getServiceGroup()
  33.     {
  34.         return $this->serviceGroup;
  35.     }
  37.     /**
  38.      * @param mixed $serviceGroup
  39.      */
  40.     public function setServiceGroup($serviceGroup): void
  41.     {
  42.         $this->serviceGroup = $serviceGroup;
  43.     }
  45.     /**
  46.      * @return mixed
  47.      */
  48.     public function getServiceAction()
  49.     {
  50.         return $this->serviceAction;
  51.     }
  53.     /**
  54.      * @param mixed $serviceAction
  55.      */
  56.     public function setServiceAction($serviceAction): void
  57.     {
  58.         $this->serviceAction = $serviceAction;
  59.         $this->initialize();
  60.     }
  62.     /**
  63.      * @return null
  64.      */
  65.     public function getArgs()
  66.     {
  67.         return $this->args;
  68.     }
  70.     /**
  71.      * @param null $args
  72.      */
  73.     public function setArgs($args): void
  74.     {
  75.         $this->args = $args;
  76.     }
  78.     protected function initialize(): void
  79.     {
  80.         if(empty($this->serviceAction)){
  81.             $this->serviceAction = 'index';
  82.         }
  83.     }
  85. }

json相应结构体

  2. namespace EasySwoole\Core\Component\Rpc\Client;
  5. use EasySwoole\Core\Component\Rpc\Server\ServiceNode;
  7. class ServiceResponse extends \EasySwoole\Core\Component\Rpc\Common\ServiceResponse
  8. {
  9.     protected $responseNode = null;
  11.     /**
  12.      * @return null
  13.      */
  14.     public function getResponseNode():?ServiceNode
  15.     {
  16.         return $this->responseNode;
  17.     }
  19.     /**
  20.      * @param null $responseNode
  21.      */
  22.     public function setResponseNode($responseNode): void
  23.     {
  24.         $this->responseNode = $responseNode;
  25.     }
  27. }

默认状态码规则

  1. namespace EasySwoole\Core\Component\Rpc\Common;
  4. class Status
  5. {
  6.     const OK = 1;//rpc调用成功
  7.     const SERVICE_REJECT_REQUEST = 0;//服务端拒绝执行，比如缺参数，或是恶意调用
  8.     const SERVICE_NOT_FOUND = -1;//服务端告诉客户端没有该服务
  9.     const SERVICE_GROUP_NOT_FOUND = -2;//服务端告诉客户端该服务不存在该服务组（服务控制器）
  10.     const SERVICE_ACTION_NOT_FOUND = -3;//服务端告诉客户端没有该action
  11.     const SERVICE_ERROR = -4;//服务端告诉客户端服务端出现了错误
  12.     const PACKAGE_ENCRYPT_DECODED_ERROR = -5;//服务端告诉客户端发过来的包openssl解密失败
  13.     const PACKAGE_DECODE_ERROR = -6;//服务端告诉客户端发过来的包无法成功解码为ServiceCaller
  14.     const CLIENT_WAIT_RESPONSE_TIMEOUT = -7;//客户端等待响应超时
  15.     const CLIENT_CONNECT_FAIL = -8;//客户端连接到服务端失败
  16.     const CLIENT_SERVER_NOT_FOUND = -9;//客户端无法找到该服务
  17. }

默认tcp协议包体规则

  1. [
  2. 'open_length_check' => true,
  3.  'package_length_type' => 'N',
  4.  'package_length_offset' => 0,
  5.  'package_body_offset' => 4,
  6.  'package_max_length' => 1024 * 64
  7. ]

PHP示例代码

  1. $opensslKey = null;
  2. $opensslMethod = 'DES-EDE3';
  4. //构造服务调用
  5. $data = [
  6.      'serviceName'=>'A',//服务名称
  7.      'serviceGroup'=>'G',//服务组（RPC服务控制器名称）
  8.      'serviceAction'=>'index',//服务行为名（RPC服务控制器action名称）
  9.      'args'=>[
  10.          'a'=>1,
  11.          'b'=>2
  12.      ]
  13. ];
  14. $fp = stream_socket_client('tcp://127.0.0.1:9502');
  15. //数据打包
  16. $sendStr = json_encode($data,JSON_UNESCAPED_UNICODE|JSON_UNESCAPED_SLASHES);
  18. if($opensslKey){
  19.     $sendStr = openssl_encrypt($sendStr,$opensslMethod,$opensslKey);
  20. }
  22. fwrite($fp,pack('N', strlen($sendStr)).$sendStr);
  23. //需要超时机制的请自己用sock time out
  24. $data = fread($fp,65533);
  25. //做长度头部校验
  26. $len = unpack('N',$data);
  27. $data = substr($data,'4');
  28. if(strlen($data) != $len[1]){
  29.     echo 'data error';
  30. }else{
  31.     if($opensslKey){
  32.         $data = openssl_decrypt($data,$opensslMethod,$opensslKey);
  33.     }
  34.     $json = json_decode($data,true);
  35.     //这就是服务端返回的结果，
  36.     var_dump($json);
  37. }
  38. fclose($fp);

NodeJs 示例代码

  1. var net = require('net');
  2. var pack = require('php-pack').pack;
  3. var unpack = require('php-pack').unpack;
  4. var json = {
  5.     serviceName:'A',
  6.     serviceGroup:'G',
  7.     serviceAction:'index',
  8.     args:[]
  9. };
  11. var send = JSON.stringify(json);
  13. send = Buffer.concat([pack("N",send.length), Buffer.from(send)]);
  15. var client = new net.Socket();
  16. client.connect(9502, '127.0.0.1', function() {
  17.     console.log('Connected');
  18.     client.write(send);
  20. });
  22. client.on('data', function(data) {
  23.     console.log('Received: ' + data);
  24.     var ret = JSON.parse(data.toString().substr(4));
  25.     console.log('status: ' +  ret.status);
  26.     client.destroy()
  27. });
  29. client.on('close', function() {
  30.     console.log('Connection closed');
  31. });
  32. client.on('error',function (error) {
  33.     console.log(error);
  34. });