找回密码
当前位置：Mikel > 开发笔记 > PHP > 正文

深入理解GatewayWorker框架 - 简书

2021-12-28 分类：PHP 阅读(799) 评论(0)

来源： 深入理解GatewayWorker框架 – 简书

序言

本文只是结合GatewayWorker和Workerman的官方文档和源码，深入了解执行过程。以便更深入的了解并使用

GatewayWorker基于Workerman开发的一个项目框架。Register进程负责保存Gateway进程和BusinessWorker进程的地址，建立两者的连接。Gateway进程负责维持客户端连接，并转发客户端的数据给BusinessWorker进程处理，BusinessWorker进程负责处理实际的业务逻辑（默认调用Events.php处理业务），并将结果推送给对应的客户端。Register、Gateway、BusinessWorker进程都是继承Worker类实现各自的功能，所以了解GatewayWorker框架的内部执行过程，需要优先理解Worker的内容

GatewayWorker目录结构

├── Applications // 这里是所有开发者应用项目
│   └── YourApp  // 其中一个项目目录，目录名可以自定义
│       ├── Events.php // 开发者只需要关注这个文件
│       ├── start_gateway.php // gateway进程启动脚本，包括端口        号等设置
│       ├── start_businessworker.php // businessWorker进程启动  脚本
│       └── start_register.php // 注册服务启动脚本
│
├── start.php // 全局启动脚本，此脚本会依次加载Applications/项目/start_*.php启动脚本
│
└── vendor    // GatewayWorker框架和Workerman框架源码目  录，此目录开发者不用关心

start.php 为启动脚本，在该脚本中，统一加载start_gateway.php start_businessworker.php start_register.php进程脚本，最后通过Worker::runAll();运行所有服务。

工作原理

1、Register、Gateway、BusinessWorker进程启动
2、Gateway、BusinessWorker进程启动后向Register服务进程发起长连接注册自己
3、Register服务收到Gateway的注册后，把所有Gateway的通讯地址保存在内存中
4、Register服务收到BusinessWorker的注册后，把内存中所有的Gateway的通讯地址发给BusinessWorker
5、BusinessWorker进程得到所有的Gateway内部通讯地址后尝试连接Gateway
6、如果运行过程中有新的Gateway服务注册到Register（一般是分布式部署加机器），则将新的Gateway内部通讯地址列表将广播给所有BusinessWorker，BusinessWorker收到后建立连接
7 、如果有Gateway下线，则Register服务会收到通知，会将对应的内部通讯地址删除，然后广播新的内部通讯地址列表给所有BusinessWorker，BusinessWorker不再连接下线的Gateway
8、至此Gateway与BusinessWorker通过Register已经建立起长连接
9、客户端的事件及数据全部由Gateway转发给BusinessWorker处理，BusinessWorker默认调用Events.php中的onConnect onMessage onClose处理业务逻辑。
10、BusinessWorker的业务逻辑入口全部在Events.php中，包括onWorkerStart进程启动事件(进程事件)、onConnect连接事件(客户端事件)、onMessage消息事件（客户端事件）、onClose连接关闭事件（客户端事件）、onWorkerStop进程退出事件（进程事件）

1 Register、Gateway、BusinessWorker进程启动

项目根目录下的start.php 为启动脚本，在该脚本中，加载start_gateway.php start_businessworker.php start_register.php进程脚本，完成各个服务的Worker初始化：

// 加载所有Applications/*/start.php，以便启动所有服务
foreach(glob(__DIR__.'/Applications/*/start*.php') as $start_file)
{
    require_once $start_file;
}

最后通过Worker::runAll();运行所有服务。

// 运行所有服务
Worker::runAll();
运行所有服务，先看一遍runAll()方法的执行内容

public static function runAll()
{
    // 检查运行环境
    self::checkSapiEnv();
    //初始化环境变量
    self::init();
    // 解析命令
    self::parseCommand();
    // 尝试以守护进程模式运行
    self::daemonize();
    // 初始化所有worker实例，主要是监听端口
    self::initWorkers();
    // 初始化所有信号处理函数
    self::installSignal();
    // 保存主进程pid
    self::saveMasterPid();
    // 展示启动界面
    self::displayUI();
    // 创建子进程(worker进程),然后给每个子进程绑定loop循环监听事件tcp
    self::forkWorkers();
    // 尝试重定向标准输入输出
    self::resetStd();
    // 监控所有子进程（worker进程）
    self::monitorWorkers();
}

self::init()初始化环境变量中，有以下部分代码，保存$_idMap从PID映射到工作进程ID

// Init data for worker id.
   self::initId();
protected static function initId()
 {
   foreach (self::$_workers as $worker_id => $worker) {
       $new_id_map = array();
       for($key = 0; $key < $worker->count; $key++) {
          $new_id_map[$key] = isset(self::$_idMap[$worker_id]      [$key]) ? self::$_idMap[$worker_id][$key] : 0;
       }
       self::$_idMap[$worker_id] = $new_id_map;
   }
 }

self::forkWorkers()方法通过循环self::$_workers数组，fork各自worker的count数量的进程。然后通过调用

$worker->run();

运行当前worker实例，在run方法中通过

  if (!self::$globalEvent) {
       $event_loop_class = self::getEventLoopName();
      self::$globalEvent = new $event_loop_class;
      // Register a listener to be notified when server socket is ready to read.
       if ($this->_socketName) {
           if ($this->transport !== 'udp') {
               self::$globalEvent->add($this->_mainSocket, EventInterface::EV_READ,
                   array($this, 'acceptConnection'));
           } else {
               self::$globalEvent->add($this->_mainSocket, EventInterface::EV_READ,
                   array($this, 'acceptUdpConnection'));
           }
       }

获取一个当前可用的事件轮询方式，然后根据当前的协议类型添加一个监听到事件轮询中
然后，尝试出发当前进程模型的onWorkerStart回调，此回调会在Gateway类以及BusinessWorker类中都会定义，代码

 if ($this->onWorkerStart) {
       try {
           call_user_func($this->onWorkerStart, $this);
       } catch (\Exception $e) {
           self::log($e);
           // Avoid rapid infinite loop exit.
           sleep(1);
           exit(250);
       } catch (\Error $e) {
           self::log($e);
           // Avoid rapid infinite loop exit.
           sleep(1);
           exit(250);
       }
   }

最后，执行事件的循环等待socket事件，处理读写等操作，代码

 // Main loop.
   self::$globalEvent->loop();

以上是runAll()方法的部分内容，会在了解GatewayWorker的工作原理的时候用到

2.1 Gateway进程向Register服务进程发起长连接注册自己

初始化Gateway

$gateway = new Gateway("text://0.0.0.0:8383");

在Gateway类中重写run方法，当调用runAll()方法启动进程时，fork进程之后，运行worker实例的时候，会调用到此重写的run方法

public function run()
{
    // 保存用户的回调，当对应的事件发生时触发
    $this->_onWorkerStart = $this->onWorkerStart;
    $this->onWorkerStart  = array($this, 'onWorkerStart');
    // 保存用户的回调，当对应的事件发生时触发
    $this->_onConnect = $this->onConnect;
    $this->onConnect  = array($this, 'onClientConnect');
    // onMessage禁止用户设置回调
    $this->onMessage = array($this, 'onClientMessage');
    // 保存用户的回调，当对应的事件发生时触发
    $this->_onClose = $this->onClose;
    $this->onClose  = array($this, 'onClientClose');
    // 保存用户的回调，当对应的事件发生时触发
    $this->_onWorkerStop = $this->onWorkerStop;
    $this->onWorkerStop  = array($this, 'onWorkerStop');
    $this->_startTime = time();
    // 运行父方法
    parent::run();
}

定义了$this->onWorkerStart回调，

$this->onWorkerStart  = array($this, 'onWorkerStart');

 

 

执行到Worker类中的run()方法时，被触发。即，上边提到的

Worker脚本中的run方法

调用Gateway类中的onWorkerStart方法，代码

public function onWorkerStart()
{
    $this->lanPort = $this->startPort + $this->id;
    if ($this->pingInterval > 0) {
        $timer_interval = $this->pingNotResponseLimit > 0 ? $this->pingInterval / 2 : $this->pingInterval;
        Timer::add($timer_interval, array($this, 'ping'));
    }
    if ($this->lanIp !== '127.0.0.1') {
        Timer::add(self::PERSISTENCE_CONNECTION_PING_INTERVAL, array($this, 'pingBusinessWorker'));
    }
    if (strpos($this->registerAddress, '127.0.0.1') !== 0) {
        Timer::add(self::PERSISTENCE_CONNECTION_PING_INTERVAL, array($this, 'pingRegister'));
    }
    if (!class_exists('\Protocols\GatewayProtocol')) {
        class_alias('GatewayWorker\Protocols\GatewayProtocol', 'Protocols\GatewayProtocol');
    }
    // 初始化 gateway 内部的监听，用于监听 worker 的连接已经连接上发来的数据
    $this->_innerTcpWorker = new Worker("GatewayProtocol://{$this->lanIp}:{$this->lanPort}");
    $this->_innerTcpWorker->listen();
    // 重新设置自动加载根目录
    Autoloader::setRootPath($this->_autoloadRootPath);
    // 设置内部监听的相关回调
    $this->_innerTcpWorker->onMessage = array($this, 'onWorkerMessage');
    $this->_innerTcpWorker->onConnect = array($this, 'onWorkerConnect');
    $this->_innerTcpWorker->onClose   = array($this, 'onWorkerClose');
    // 注册 gateway 的内部通讯地址，worker 去连这个地址，以便 gateway 与 worker 之间建立起 TCP 长连接
    $this->registerAddress();
    if ($this->_onWorkerStart) {
        call_user_func($this->_onWorkerStart, $this);
    }
}

$this->startPort : 内部通讯起始端口，假如$gateway->count=4，起始端口为4000，可在gateway启动脚本中自定义
$this->id : 基于worker实例分配的进程编号，当前从0开始，根据count自增。在fork进程的时候生成

Worker.php

$this->_innerTcpWorker：用于监听 worker 的连接已经连接上发来的数据。在工作原理5中，BusinessWorker进程得到所有的Gateway内部通讯地址后尝试连接Gateway以及其他两者之间的通信(连接，消息，关闭)会被调用
$this->registerAddress()： 代码中$this->registerAddress是在start_gateway.php初始化Gateway类之后定义的。该端口是Register进程所监听。此处异步的向Register进程发送数据，存储当前 Gateway 的内部通信地址 

public function registerAddress()
{
    $address                   = $this->lanIp . ':' . $this->lanPort;
    $this->_registerConnection = new AsyncTcpConnection("text://{$this->registerAddress}");
    $this->_registerConnection->send('{"event":"gateway_connect", "address":"' . $address . '", "secret_key":"' . $this->secretKey . '"}');
    $this->_registerConnection->onClose = array($this, 'onRegisterConnectionClose');
    $this->_registerConnection->connect();
}

$this->lanIp: Gateway所在服务器的内网IP

2.2 BusinessWorker进程向Register服务进程发起长连接注册自己

BusinessWorker类中同样重写run方法，定义了$this->onWorkerStart

 public function run()
 {
    $this->_onWorkerStart  = $this->onWorkerStart;
    $this->_onWorkerReload = $this->onWorkerReload;
    $this->_onWorkerStop = $this->onWorkerStop;
    $this->onWorkerStop   = array($this, 'onWorkerStop');
    $this->onWorkerStart   = array($this, 'onWorkerStart');
    $this->onWorkerReload  = array($this, 'onWorkerReload');
    parent::run();
 }

执行Worker类中的run方法，触发BusinessWorker中的onWorkerStart

protected function onWorkerStart()
{
    if (!class_exists('\Protocols\GatewayProtocol')) {
        class_alias('GatewayWorker\Protocols\GatewayProtocol', 'Protocols\GatewayProtocol');
    }
    $this->connectToRegister();
    \GatewayWorker\Lib\Gateway::setBusinessWorker($this);
    \GatewayWorker\Lib\Gateway::$secretKey = $this->secretKey;
    if ($this->_onWorkerStart) {
        call_user_func($this->_onWorkerStart, $this);
    }
    
    if (is_callable($this->eventHandler . '::onWorkerStart')) {
        call_user_func($this->eventHandler . '::onWorkerStart', $this);
    }

    if (function_exists('pcntl_signal')) {
        // 业务超时信号处理
        pcntl_signal(SIGALRM, array($this, 'timeoutHandler'), false);
    } else {
        $this->processTimeout = 0;
    }

    // 设置回调
    if (is_callable($this->eventHandler . '::onConnect')) {
        $this->_eventOnConnect = $this->eventHandler . '::onConnect';
    }

    if (is_callable($this->eventHandler . '::onMessage')) {
        $this->_eventOnMessage = $this->eventHandler . '::onMessage';
    } else {
        echo "Waring: {$this->eventHandler}::onMessage is not callable\n";
    }

    if (is_callable($this->eventHandler . '::onClose')) {
        $this->_eventOnClose = $this->eventHandler . '::onClose';
    }

    // 如果Register服务器不在本地服务器，则需要保持心跳
    if (strpos($this->registerAddress, '127.0.0.1') !== 0) {
        Timer::add(self::PERSISTENCE_CONNECTION_PING_INTERVAL, array($this, 'pingRegister'));
    }
}

通过connectToRegister方法，发送数据到Register进程，连接服务注册中心

public function connectToRegister()
{
    $this->_registerConnection = new AsyncTcpConnection("text://{$this->registerAddress}");
    $this->_registerConnection->send('{"event":"worker_connect","secret_key":"' . $this->secretKey . '"}');
    $this->_registerConnection->onClose   = array($this, 'onRegisterConnectionClose');
    $this->_registerConnection->onMessage = array($this, 'onRegisterConnectionMessage');
    $this->_registerConnection->connect();
}

3 Register服务收到Gateway的注册后，把所有的Gateway的通讯地址保存在内存中

在Register类中，重写了run方法，定义了当前的

     $this->onConnect = array($this, 'onConnect');
    // 设置 onMessage 回调
    $this->onMessage = array($this, 'onMessage');

    // 设置 onClose 回调
    $this->onClose = array($this, 'onClose');

三个属性，当Register启动的进程收到消息时，会触发onMessage方法

 public function onMessage($connection, $buffer)
{
    // 删除定时器
    Timer::del($connection->timeout_timerid);
    $data       = @json_decode($buffer, true);
    if (empty($data['event'])) {
        $error = "Bad request for Register service. Request info(IP:".$connection->getRemoteIp().", Request Buffer:$buffer). See http://wiki.workerman.net/Error4 for detail";
        Worker::log($error);
        return $connection->close($error);
    }
    $event      = $data['event'];
    $secret_key = isset($data['secret_key']) ? $data['secret_key'] : '';
    // 开始验证
    switch ($event) {
        // 是 gateway 连接
        case 'gateway_connect':
            if (empty($data['address'])) {
                echo "address not found\n";
                return $connection->close();
            }
            if ($secret_key !== $this->secretKey) {
                Worker::log("Register: Key does not match ".var_export($secret_key, true)." !== ".var_export($this->secretKey, true));
                return $connection->close();
            }
            $this->_gatewayConnections[$connection->id] = $data['address'];
            $this->broadcastAddresses();
            break;
        // 是 worker 连接
        case 'worker_connect':
            if ($secret_key !== $this->secretKey) {
                Worker::log("Register: Key does not match ".var_export($secret_key, true)." !== ".var_export($this->secretKey, true));
                return $connection->close();
            }
            $this->_workerConnections[$connection->id] = $connection;
            $this->broadcastAddresses($connection);
            break;
        case 'ping':
            break;
        default:
            Worker::log("Register unknown event:$event IP: ".$connection->getRemoteIp()." Buffer:$buffer. See http://wiki.workerman.net/Error4 for detail");
            $connection->close();
    }
}

当$event = ‘gateway_connect’时，是Gateway发来的注册消息，保存到$this->_gatewayConnections数组中，在通过broadcastAddresses方法将当前$this->_gatewayConnections中所有的Gatewat通讯地址转发给所有BusinessWorker进程

4 Register服务收到BusinessWorker的注册后，把内存中所有的Gateway的通讯地址发给BusinessWorker

 

 

同第3步中，Register类收到BusinessWorker的注册时，会触发onMessage方法中的worker_connect，case选项。

image.png

同时，将当前worker连接加入到$_workerConnections数组中，在通过broadcastAddresses方法将当前$this->_gatewayConnections中所有的Gatewat通讯地址转发给所有BusinessWorker进程。

5 BusinessWorker进程得到所有的Gateway内部通讯地址后尝试连接Gateway

在BusinessWoker类的启动中，通过重写run方法，定义的启动onWorkerStart方法中，通过connectToRegister方法注册服务中心的同时，也定义了onMessage匿名函数，用于接收消息回调。

$this->_registerConnection->onMessage = array($this, 'onRegisterConnectionMessage');

即，当注册中心发来消息时候，回调到此处

 public function onRegisterConnectionMessage($register_connection, $data)
{
    $data = json_decode($data, true);
    if (!isset($data['event'])) {
        echo "Received bad data from Register\n";
        return;
    }
    $event = $data['event'];
    switch ($event) {
        case 'broadcast_addresses':
            if (!is_array($data['addresses'])) {
                echo "Received bad data from Register. Addresses empty\n";
                return;
            }
            $addresses               = $data['addresses'];
            $this->_gatewayAddresses = array();
            foreach ($addresses as $addr) {
                $this->_gatewayAddresses[$addr] = $addr;
            }
            $this->checkGatewayConnections($addresses);
            break;
        default:
            echo "Receive bad event:$event from Register.\n";
    }
}

其中Register类发来的数据是

$data   = array(
        'event'     => 'broadcast_addresses',
        'addresses' => array_unique(array_values($this->_gatewayConnections)),
    );

这个时候，就会通过checkGatewayConnections方法检查gateway的这些通信端口是否都已经连接，在通过tryToConnectGateway方法尝试连接gateway的这些内部通信地址

6 Gateway进程收到BusinessWorker进程的连接消息

同样，在Gateway进程启动的时候，触发的onWorkerStart方法中，也定义了一个内部通讯的onWorkerMessage

$this->_innerTcpWorker->onMessage = array($this, 'onWorkerMessage');

由此来接收BusinessWorker进程发来的连接消息，部分代码

public function onWorkerMessage($connection, $data)
{
    $cmd = $data['cmd'];
    if (empty($connection->authorized) && $cmd !== GatewayProtocol::CMD_WORKER_CONNECT && $cmd !== GatewayProtocol::CMD_GATEWAY_CLIENT_CONNECT) {
        self::log("Unauthorized request from " . $connection->getRemoteIp() . ":" . $connection->getRemotePort());
        return $connection->close();
    }
    switch ($cmd) {
        // BusinessWorker连接Gateway
        case GatewayProtocol::CMD_WORKER_CONNECT:
            $worker_info = json_decode($data['body'], true);
            if ($worker_info['secret_key'] !== $this->secretKey) {
                self::log("Gateway: Worker key does not match ".var_export($this->secretKey, true)." !== ". var_export($this->secretKey));
                return $connection->close();
            }
            $key = $connection->getRemoteIp() . ':' . $worker_info['worker_key'];
            // 在一台服务器上businessWorker->name不能相同
            if (isset($this->_workerConnections[$key])) {
                self::log("Gateway: Worker->name conflict. Key:{$key}");
        $connection->close();
                return;
            }
    $connection->key = $key;
            $this->_workerConnections[$key] = $connection;
            $connection->authorized = true;
            return;
        // GatewayClient连接Gateway

将worker的进程连接保存到$this->_workerConnections[$key] = $connection;

7 Gateway进程收到客户端的连接，消息时，会通过Gateway转发给worker处理

 // Gateway类的run方法中定义此属性
 $this->onMessage = array($this, 'onClientMessage');
 
 // 收到客户端消息的时候出发此函数
 public function onClientMessage($connection, $data)
 {
    $connection->pingNotResponseCount = -1;
    $this->sendToWorker(GatewayProtocol::CMD_ON_MESSAGE, $connection, $data);
 }

在sendToWorker方法中，将数据发给worker进程处理

作者：AntFoot
链接：https://www.jianshu.com/p/47a9f2f0c18b
来源：简书
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。

赞(0) 打赏
分享到： 更多 (0)
标签：

相关推荐

热门标签

C# (810)ASP.NET MVC (618)Android (587)ASP.NET (447)数据库 (299)JavaScript (285)JQuery (221)PHP (209)SQLServer (197)程序开发 (156)架构设计 (155)源价值 (155)SQL (128)Java (111)EasyUi (85)教程 (76)网站 (67)Debug (63)互联网 (63)MySQL (59)Flash (56)搜索优化 (54)创意 (53)SEO (51)ASP.NET MVC3 (51)电子商务 (50)建站经验 (49)实例代码 (49)源码下载 (47)原创 (45)

分类目录

链接表

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

微信扫一扫打赏