EasySwoole RPC 自定义注册中心

EasySwoole 默认为通过UDP广播+自定义进程定时刷新自身节点信息的方式来实现无主化/注册中心的服务发现。在服务正常关闭的时候，自定义定时进程的onShutdown方法会执行deleteServiceNode 方法来实现节点下线。在非正常关闭的时候，心跳超时也会被节点管理器踢出， 有些情况，不方便用UDP广播的情况下，那么EasySwoole支持你自定义一个节点管理器，来变更服务发现方式。

例如用Redis来实现

  1. namespace EasySwoole\Rpc\NodeManager;
  3. use EasySwoole\RedisPool\RedisPool;
  4. use EasySwoole\Rpc\ServiceNode;
  5. use EasySwoole\Utility\Random;
  7. class RedisManager implements NodeManagerInterface
  8. {
  9.     protected $redisKey;
  11.     protected $pool;
  13.     function __construct(RedisPool $pool, string $hashKey = 'rpc')
  14.     {
  15.         $this->redisKey = $hashKey;
  16.         $this->pool = $pool;
  17.     }
  19.     function getServiceNodes(string $serviceName, ?string $version = null): array
  20.     {
  21.         $redis = $this->pool->getObj(15);
  22.         try {
  23.             $nodes = $redis->hGetAll("{$this->redisKey}_{$serviceName}");
  24.             $nodes = $nodes ?: [];
  25.             $ret = [];
  26.             foreach ($nodes as $nodeId => $node) {
  27.                 $node = new ServiceNode(json_decode($node,true));
  28.                 /**
  29.                  * @var  $nodeId
  30.                  * @var  ServiceNode $node
  31.                  */
  32.                 if (time() - $node->getLastHeartBeat() > 30) {
  33.                     $this->deleteServiceNode($node);
  34.                 }
  35.                 if ($version && $version != $node->getServiceVersion()) {
  36.                     continue;
  37.                 }
  38.                 $ret[$nodeId] = $node;
  39.             }
  40.             return $ret;
  41.         } catch (\Throwable $throwable) {
  42.             //如果该redis断线则销毁
  43.             $this->pool->unsetObj($redis);
  44.         } finally {
  45.             $this->pool->recycleObj($redis);
  46.         }
  47.         return [];
  48.     }
  50.     function getServiceNode(string $serviceName, ?string $version = null): ?ServiceNode
  51.     {
  52.         $list = $this->getServiceNodes($serviceName, $version);
  53.         if (empty($list)) {
  54.             return null;
  55.         }
  56.         return Random::arrayRandOne($list);
  57.     }
  59.     function deleteServiceNode(ServiceNode $serviceNode): bool
  60.     {
  61.         $redis = $this->pool->getObj(15);
  62.         try {
  63.             $redis->hDel($this->generateNodeKey($serviceNode), $serviceNode->getNodeId());
  64.             return true;
  65.         } catch (\Throwable $throwable) {
  66.             $this->pool->unsetObj($redis);
  67.         } finally {
  68.             $this->pool->recycleObj($redis);
  69.         }
  70.         return false;
  71.     }
  73.     function serviceNodeHeartBeat(ServiceNode $serviceNode): bool
  74.     {
  75.         if (empty($serviceNode->getLastHeartBeat())) {
  76.             $serviceNode->setLastHeartBeat(time());
  77.         }
  78.         $redis = $this->pool->getObj(15);
  79.         try {
  80.             $redis->hSet($this->generateNodeKey($serviceNode), $serviceNode->getNodeId(), $serviceNode->__toString());
  81.             return true;
  82.         } catch (\Throwable $throwable) {
  83.             $this->pool->unsetObj($redis);
  84.         } finally {
  85.             //这边需要测试一个对象被unset后是否还能被回收
  86.             $this->pool->recycleObj($redis);
  87.         }
  88.         return false;
  89.     }
  91.     protected function generateNodeKey(ServiceNode $node)
  92.     {
  93.         return "{$this->redisKey}_{$node->getServiceName()}";
  94.     }
  95. }

::: warning 即使关闭了UDP定时广,EasySwoole Rpc的tick进程依旧会每3秒执行一次serviceNodeHeartBeat用于更新自身的节点心跳信息。 :::