登录
目录
当前位置： 首页 > 文档资料 > easySwoole 2.x 中文文档 >

系统组件 - Cluster

优质
小牛编辑
138浏览
2023-12-01

EasySwoole 提供基础的对等模式分布式通讯支持。

知识储备

UDP

什幺是UDP协议

什幺是UDP协议请自行百度。

UDP广播地址

广播地址(Broadcast Address)是专门用于同时向网络中所有工作站进行发送的一个地址。在使用TCP/IP 协议的网络中，主机标识段host ID 为全1 的IP 地址为广播地址，广播的分组传送给host ID段所涉及的所有计算机。例如，对于10.1.1.0 （255.255.255.0 ）网段，其广播地址为10.1.1.255 （255 即为2 进制的11111111 ），当发出一个目的地址为10.1.1.255 的分组（封包）时，它将被分发给该网段上的所有计算机。

实现讲解

原理讲解

在开启集群模式的时候，EasySwoole会开启一个UDP子服务，用于集群的UDP信息广播，每当有收到信息，就会自动进行openssl解密，并根据解析后的Message，
执行Message回调容器中与Message Command字段匹配的事件。在BaseService的自定义进程中，每一个信息广播周期，都会执行广播回调容器内的全部事件，而EasySwoole注册了默认的集群节点广播、RPC广播（全部消息都openssl加密）。

核心代码讲解

Cluster.php

  1. <?php
  2. /**
  3.  * Created by PhpStorm.
  4.  * User: yf
  5.  * Date: 2018/1/24
  6.  * Time: 下午10:51
  7.  */
  9. namespace EasySwoole\Core\Component\Cluster;
  12. use EasySwoole\Config;
  13. use EasySwoole\Core\AbstractInterface\Singleton;
  14. use EasySwoole\Core\Component\Cluster\Callback\BroadcastCallbackContainer;
  15. use EasySwoole\Core\Component\Cluster\Callback\DefaultCallbackName;
  16. use EasySwoole\Core\Component\Cluster\Callback\ShutdownCallBackContainer;
  17. use EasySwoole\Core\Component\Cluster\Common\BaseServiceProcess;
  18. use EasySwoole\Core\Component\Cluster\Common\MessageBean;
  19. use EasySwoole\Core\Component\Cluster\Common\NodeBean;
  20. use EasySwoole\Core\Component\Cluster\Callback\MessageCallbackContainer;
  21. use EasySwoole\Core\Component\Cluster\NetWork\Deliverer;
  22. use EasySwoole\Core\Component\Cluster\NetWork\PacketParser;
  23. use EasySwoole\Core\Component\Openssl;
  24. use EasySwoole\Core\Component\Rpc\Common\ServiceNode;
  25. use EasySwoole\Core\Component\Rpc\Server;
  26. use EasySwoole\Core\Component\Trigger;
  27. use EasySwoole\Core\Socket\Client\Udp;
  28. use EasySwoole\Core\Swoole\EventHelper;
  29. use EasySwoole\Core\Swoole\Memory\TableManager;
  30. use EasySwoole\Core\Swoole\Process\ProcessManager;
  31. use EasySwoole\Core\Swoole\ServerManager;
  32. use Swoole\Table;
  34. class Cluster
  35. {
  36.     use Singleton;
  38.     private $currentNode;
  40.     function __construct()
  41.     {
  42.         $conf = Config::getInstance()->getConf('CLUSTER');
  43.         //加载配置信息，并实例化当前节点
  44.         $this->currentNode = new NodeBean($conf);
  45.         if($this->currentNode->getEnable() && empty($this->currentNode->getToken())){
  46.             Trigger::throwable(new \Exception('cluster token could not be empty and set cluster mode disable automatic'));
  47.             $this->currentNode->setEnable(false) ;
  48.         }
  49.         if($this->currentNode->getEnable() && empty($this->currentNode->getListenAddress())){
  50.             Trigger::throwable(new \Exception('cluster listenAddress could not be empty and set cluster mode disable automatic'));
  51.             $this->currentNode->setEnable(false) ;
  52.         }
  53.         //初始化swoole table用于记录节点信息
  54.         TableManager::getInstance()->add('ClusterNodeList',[
  55.             'nodeName'=>[
  56.                 'type'=>Table::TYPE_STRING,'size'=>20
  57.             ],
  58.             'udpAddress'=>[
  59.                 'type'=>Table::TYPE_STRING,'size'=>16
  60.             ],
  61.             'udpPort'=>[
  62.                 'type'=>Table::TYPE_INT,'size'=>10
  63.             ],
  64.             'listenPort'=>[
  65.                 'type'=>Table::TYPE_STRING,'size'=>10
  66.             ],
  67.             'lastBeatBeatTime'=>[
  68.                 'type'=>Table::TYPE_INT,'size'=>10
  69.             ]
  70.         ]);
  71.     }
  73.     function run()
  74.     {
  75.         if($this->currentNode->getEnable()){
  76.             //注册默认回调
  77.             self::registerDefaultCallback();
  78.             $name = Config::getInstance()->getConf('SERVER_NAME');
  79.             //注册基础服务进程
  80.             ProcessManager::getInstance()->addProcess("{$name}_Cluster_BaseService",BaseServiceProcess::class,['currentNode'=>$this->currentNode]);
  81.             //开启UDP子服务，
  82.             $sub = ServerManager::getInstance()->addServer("{$name}_Cluster",$this->currentNode->getListenPort(),SWOOLE_SOCK_UDP,$this->currentNode->getListenAddress());
  83.             $openssl = new Openssl($this->currentNode->getToken());
  84.             EventHelper::register($sub,$sub::onPacket,function (\swoole_server $server, string $data, array $client_info)use($openssl){
  85.                 $data = $openssl->decrypt($data);
  86.                 $udpClient = new Udp($client_info);
  87.                 //解析信息包并执行回调
  88.                 $message = PacketParser::unpack((string)$data,$udpClient);
  89.                 if($message){
  90.                     MessageCallbackContainer::getInstance()->hook($message->getCommand(),$message);
  91.                 }
  92.             });
  93.         }
  94.     }
  96.     function currentNode():NodeBean
  97.     {
  98.         return $this->currentNode;
  99.     }
  101.     function allNodes():array
  102.     {
  103.         $ret = [];
  104.         $list = TableManager::getInstance()->get('ClusterNodeList');
  105.         $time = time();
  106.         $ttl = $this->currentNode->getNodeTimeout();
  107.         foreach ($list as $key => $item){
  108.             $node = new NodeBean([
  109.                 'nodeId'=>$key,
  110.                 'nodeName'=>$item['nodeName'],
  111.                 'lastBeatBeatTime'=>$item['lastBeatBeatTime'],
  112.                 'udpInfo'=>[
  113.                     'address'=>$item['udpAddress'],
  114.                     'port'=>$item['udpPort']
  115.                 ],
  116.                 'listenPort'=>$item['listenPort']
  117.             ]);
  118.             if($time - $item['lastBeatBeatTime'] > $ttl){
  119.                 //异常下线
  120.                 NodeOffLienCallbackContainer::getInstance()->call($node,false);
  121.                 TableManager::getInstance()->get('ClusterNodeList')->del($key);
  122.             }else{
  123.                 $ret[] = $node;
  124.             }
  125.         }
  126.         return $ret;
  127.     }
  129.     function getNode($nodeId):?NodeBean
  130.     {
  131.         $item = TableManager::getInstance()->get('ClusterNodeList')->get($nodeId);
  132.         if(is_array($item)){
  133.             $ttl = $this->currentNode->getNodeTimeout();
  134.             $node = new NodeBean([
  135.                 'nodeId'=>$nodeId,
  136.                 'nodeName'=>$item['nodeName'],
  137.                 'lastBeatBeatTime'=>$item['lastBeatBeatTime'],
  138.                 'udpInfo'=>[
  139.                     'address'=>$item['udpAddress'],
  140.                     'port'=>$item['udpPort']
  141.                 ],
  142.                 'listenPort'=>$item['listenPort']
  143.             ]);
  144.             if(time() - $item['lastBeatBeatTime'] > $ttl){
  145.                 NodeOffLienCallbackContainer::getInstance()->call($node,false);
  146.                 return null;
  147.             }else{
  148.                 return $node;
  149.             }
  150.         }else{
  151.             return null;
  152.         }
  153.     }
  154.     /*
  155.      * 注册默认服务
  156.      */
  157.     private static function registerDefaultCallback()
  158.     {
  159.         //集群节点广播回调
  160.         MessageCallbackContainer::getInstance()->add(DefaultCallbackName::CLUSTER_NODE_BROADCAST,function (MessageBean $messageBean){
  161.             $node = $messageBean->getFromNode();
  162.             TableManager::getInstance()->get('ClusterNodeList')->set($node->getNodeId(),[
  163.                 'nodeName'=>$node->getNodeName(),
  164.                 'udpAddress'=>$node->getUdpInfo()->getAddress(),
  165.                 'udpPort'=>$node->getUdpInfo()->getPort(),
  166.                 'lastBeatBeatTime'=>time(),
  167.                 'listenPort'=>$node->getListenPort()
  168.             ]);
  169.         });
  170.         //集群节点广播关机回调
  171.         MessageCallbackContainer::getInstance()->add(DefaultCallbackName::CLUSTER_NODE_SHUTDOWN,function (MessageBean $messageBean){
  172.             $node = $messageBean->getFromNode();
  173.             TableManager::getInstance()->get('ClusterNodeList')->del($node->getNodeId());
  174.             //下线该服务的全部rpc服务
  175.             Server::getInstance()->serverNodeOffLine($node);
  176.             //正常下线
  177.             NodeOffLienCallbackContainer::getInstance()->call($node,true);
  178.         });
  179.         //RPC服务节点广播回调
  180.         MessageCallbackContainer::getInstance()->add(DefaultCallbackName::RPC_SERVICE_BROADCAST,function (MessageBean $messageBean){
  181.             $node = $messageBean->getFromNode();
  182.             $list = $messageBean->getArgs();
  183.             foreach ($list as $item){
  184.                 $serviceNode = new ServiceNode($item);
  185.                 //可达主机地址即为udp地址（真实地址）
  186.                 $serviceNode->setAddress($node->getUdpInfo()->getAddress());
  187.                 Server::getInstance()->updateServiceNode($serviceNode);
  188.             }
  189.         });
  191.         //集群节点广播
  192.         BroadcastCallbackContainer::getInstance()->set(DefaultCallbackName::CLUSTER_NODE_BROADCAST,function (){
  193.             $message = new MessageBean();
  194.             $message->setCommand(DefaultCallbackName::CLUSTER_NODE_BROADCAST);
  195.             Deliverer::broadcast($message);
  196.         });
  197.         //RPC服务广播
  198.         BroadcastCallbackContainer::getInstance()->set(DefaultCallbackName::RPC_SERVICE_BROADCAST,function (){
  199.             $ret = Server::getInstance()->allLocalServiceNodes();
  200.             $data = [];
  201.             foreach ($ret as $item){
  202.                 $data[] = $item->toArray();
  203.             }
  204.             $message = new MessageBean();
  205.             $message->setArgs($data);
  206.             $message->setCommand(DefaultCallbackName::RPC_SERVICE_BROADCAST);
  207.             Deliverer::broadcast($message);
  208.         });
  209.         //注册默认集群关机回调
  210.         ShutdownCallBackContainer::getInstance()->set(DefaultCallbackName::CLUSTER_NODE_SHUTDOWN,function (){
  211.             $message = new MessageBean();
  212.             $message->setCommand(DefaultCallbackName::CLUSTER_NODE_SHUTDOWN);
  213.             Deliverer::broadcast($message);
  214.         });
  215.     }
  217. }

BaseServiceProcess.php

  1. namespace EasySwoole\Core\Component\Cluster\Common;
  4. use EasySwoole\Core\Component\Cluster\Callback\BroadcastCallbackContainer;
  5. use EasySwoole\Core\Component\Cluster\Callback\ShutdownCallbackContainer;
  6. use EasySwoole\Core\Swoole\Process\AbstractProcess;
  7. use Swoole\Process;
  9. class BaseServiceProcess extends AbstractProcess
  10. {
  12.     public function run(Process $process)
  13.     {
  14.         // TODO: Implement run() method.
  15.         //定时执行广播周期回调事件
  16.         $this->addTick($this->getArg('currentNode')->getBroadcastTTL()*1000,function (){
  17.             BroadcastCallbackContainer::getInstance()->call();
  18.         });
  19.     }
  21.     public function onShutDown()
  22.     {
  23.         // TODO: Implement onShutDown() method.
  24.         //守护模式的时候，正常关闭服务，会执行该回调
  25.         ShutdownCallbackContainer::getInstance()->call();
  26.     }
  28.     public function onReceive(string $str, ...$args)
  29.     {
  30.         // TODO: Implement onReceive() method.
  31.     }
  32. }

Deliverer.php 可以在任意位置，做UDP信息发送操作

  1. namespace EasySwoole\Core\Component\Cluster\NetWork;
  4. use EasySwoole\Core\Component\Cluster\Cluster;
  5. use EasySwoole\Core\Component\Cluster\Common\MessageBean;
  6. use EasySwoole\Core\Component\Cluster\Common\NodeBean;
  8. class Deliverer
  9. {
  10.     /*
  11.      * 调用此方法，请确保知晓节点的udp信息
  12.      */
  13.     public static function toNode(MessageBean $message,NodeBean $node)
  14.     {
  15.         $message = PacketParser::pack($message);
  16.         //端口以监听地址为准，ip地址以udp地址为准
  17.         Udp::sendTo($message,$node->getListenPort(),$node->getUdpInfo()->getAddress());
  18.     }
  21.     public static function toAllNode(MessageBean $message)
  22.     {
  23.         $message = PacketParser::pack($message);
  24.         $nodes = Cluster::getInstance()->allNodes();
  25.         foreach ($nodes as $node){
  26.             Udp::sendTo($message,$node->getListenPort(),$node->getUdpInfo()->getAddress());
  27.         }
  28.     }
  30.     public static function broadcast(MessageBean $message)
  31.     {
  32.         $message = PacketParser::pack($message);
  33.         $addresses = Cluster::getInstance()->currentNode()->getBroadcastAddress();
  34.         foreach ($addresses as $item){
  35.             $item = explode(':',$item);
  36.             Udp::broadcast($message,$item[1],$item[0]);
  37.         }
  38.     }
  39. }

常见问题

如何实现跨机房通讯？

假设你在阿里云和腾讯云都各有一台机器，那幺如何实现两台机器互联呢？那幺请注意，EasySwoole允许设置多个广播地址，你可以设置
[ip1:port,ip2:port],注意，此刻ip1,ip2全部为公网ip，且都开放了端口。如果是全部都在同一机房内，那幺仅需向255.255.255.255（默认）广播即可。具体的请以实际的路由网关规则为准。

某个节点异常下线怎幺办

正常情况下，当有节点下线的时候，会触发默认注册的节点下线回调事件，全部节点都会收到节点下线通知，若某台机器意外关机，那幺最大在经过配置项中的nodeTimeout以后，那幺该节点会被清除，并执行回调。

如何注册自己的命令

EasySwoole\Core\Component\Cluster\Callback路径下的全部容器，都是单利模式，主进程中，在EasySwooleEvent.php的主服务创建事件中即可注册。

如何用对等模式实现主从模式功能

假设A-Z台机器，你需要A去收集其他的机器信息，那幺你可以在A中部署独有服务，然后其他机器发送错误的时候，通过RPC,或者是UDP发送的方式，发送给A节点。

免责声明： 以上内容版权归原作者所有，如有侵犯您的原创版权请告知，我们将尽快删除相关内容。感谢每一位辛勤著写的作者，感谢每一位的分享。
@小牛知识库 xnip.cn