(4条消息)深入理解GatewayWorker框架 - houzhyan-博客 - CSDN博客
原文地址: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的通讯地址发给BusinessWorker5、BusinessWorker进程得到所有的Gateway内部通讯地址后尝试连接Gateway6、如果运行过程中有新的Gateway服务注册到Register（一般是分布式部署加机器），则将新的Gateway内部通讯地址列表将广播给所有BusinessWorker，BusinessWorker收到后建立连接7 、如果有Gateway下线，则Register服务会收到通知，会将对应的内部通讯地址删除，然后广播新的内部通讯地址列表给所有BusinessWorker，BusinessWorker不再连接下线的Gateway8、至此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进程处理
原网址: 访问
创建于: 2019-04-06 22:48:28
目录: default
标签: 无
未标明原创文章均为采集，版权归作者所有，转载无需和我联系，请注明原出处，南摩阿彌陀佛，知识，不只知道，要得到
最新评论