深入理解GatewayWorker框架

￡神魔★判官ぃ 2022-04-13 03:55 337阅读 0赞

### 原文地址:[http://www.php-master.com/post/342621.html][http_www.php-master.com_post_342621.html] ###

### 序言 ###

本文只是结合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()方法时，被触发。即，上边提到的

![1765825-80a21f906bb1acf9.png][]

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进程的时候生成

![1765825-9df0980dcd09a7a7.png][]

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选项。

![1765825-941cbd82918c6bc9.png][]

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进程处理

[http_www.php-master.com_post_342621.html]: http://www.php-master.com/post/342621.html
[1765825-80a21f906bb1acf9.png]: /images/20220413/2bdf960602a74d06ac59c32f965a5c9b.png
[1765825-9df0980dcd09a7a7.png]: /images/20220413/8ba8fc7e6f5843858924fabc4784c83a.png
[1765825-941cbd82918c6bc9.png]: /images/20220413/90c9a9e9f189453b825e0f71d9a3a517.png