另：Worker功能函數 · workerman.php源代碼閱讀

[TOC] * * * * * ## 1 成員列表 ### 1 成員屬性 ~~~ ;版本號 const VERSION = '3.3.1'; ;啟動狀態，運行狀態,停止狀態，重載狀態 const STATUS_STARTING = 1; const STATUS_RUNNING = 2; const STATUS_SHUTDOWN = 4; const STATUS_RELOADING = 8; ;子進程強制關閉時間,默認baclog長度,udp數據包最大 const KILL_WORKER_TIMER_TIME = 2; const DEFAUL_BACKLOG = 1024; const MAX_UDP_PACKAGE_SIZE = 65535; ;進程id編號,進程名稱,worker進程數量 public $id = 0; public $name = 'none'; public $count = 1; ;進程用戶，進程組 public $user = ''; public $group = ''; ;是否可以重載，是否復用端口 public $reloadable = true; public $reusePort = false; ;worker 3個回調接口 public $onWorkerStart = null; public $onWorkerStop = null; public $onWorkerReload = null; ;connect 6個回調接口 public $onConnect = null; public $onMessage = null; public $onClose = null; public $onError = null; public $onBufferFull = null; public $onBufferDrain = null; ;傳輸層協議,應用層協議 public $transport = 'tcp'; public $protocol = ''; ;所有連接 public $connections = array(); ;自動加載根目錄 protected $_autoloadRootPath = ''; ;是否守護進程模式 public static $daemonize = false; ;輸出文件,pid文件，日志文件 public static $stdoutFile = '/dev/null'; public static $pidFile = ''; public static $logFile = ''; ; 全局事件循環 public static $globalEvent = null; ; 主進程id protected static $_masterPid = 0; ; 監聽socket protected $_mainSocket = null; ; socket名稱,socket上下文選項 protected $_socketName = ''; protected $_context = null; ;workers實例數組，worker進程id數組 protected static $_workers = array(); protected static $_pidMap = array(); ;等待重啟worker進程數組，worker的pid與進程編號映射 protected static $_pidsToRestart = array(); protected static $_idMap = array(); ;當前狀態 protected static $_status = self::STATUS_STARTING; ;workername最大長度,socketname最大長度,username最大長度 protected static $_maxWorkerNameLength = 12; protected static $_maxSocketNameLength = 12; protected static $_maxUserNameLength = 12; ;狀態文件,啟動文件 protected static $_statisticsFile = ''; protected static $_startFile = ''; ;worker進程狀態信息格式， protected static $_globalStatistics ;可選事件循環,當前事件循環名稱 protected static $_availableEventLoops protected static $_eventLoopName ;內置協議 protected static $_builtinTransports ~~~ ### 2 成員函數 ~~~ 1 setProcessTitle() 設置進程名稱 2 initId() 初始化$_idMap 3 log() 日志記錄 4 getSocketName() 獲取socket名稱 5 getCurrentUser() 獲取當前用戶 6 listen() 啟動監聽端口 7 signalHandler() 信號處理函數 8 stopAll() 停止所有 9 reload() 重載 10 writeStatisticsToStatusFile() 狀態信息 11 getAllWorkerPids()　獲取worker的pid 12 forkOneWorker() 創建一個worker進程 13 getId() 獲取workerid 14 setUserAndGroup() 設置用戶信息 15 run() worker啟動 16 getEventLoopName() 獲取事件循環名稱 17 reinstallSignal() worker信號處理注冊 18 acceptConnection() 創建一個連接 19 acceptUdpConnection() upd數據包 20 stop() worker子進程關閉 ~~~ ## 2 函數分析 ### 1 setProcessTitle() `protected static function setProcessTitle($title) 設置進程名稱` > $title:進程名稱調用cli_set_process_title()或者setproctitle() ### 2 initId() `protected static function initId() 初始化$_idMap` 將workers的pid與workers的進程編號關聯到$_idMap ### 3 log() `protected static function log($msg) 日志信息` > $msg:待記錄信息 debug模式直接輸出 daemoniz模式輸出到日志文件 ### 4 getSocketName() `public function getSocketName() 獲取socketname` 將第一個字母小寫返回，或者返回none ### 5 getCurrentUser() ~~~ protected static function getCurrentUser() 當前進程的用戶 ~~~ 調用posix_getpwuid() 獲取當前進程的用戶信息 ### 6 listen()[重點] `public function listen() 啟動端口監聽` ~~~ if (!$this->_socketName || $this->_mainSocket) { return; } ~~~ 檢查_socketName與_mainSocket參數。 ~~~ Autoloader::setRootPath($this->_autoloadRootPath); ~~~ 注冊自動加載根目錄 ~~~ $local_socket = $this->_socketName; list($scheme, $address) = explode(':', $this->_socketName, 2); ~~~ 解析$_socketName為協議$scheme,和監聽地址$address ~~~ if (!isset(self::$_builtinTransports[$scheme])) { $scheme = ucfirst($scheme); $this->protocol = '\\Protocols\\' . $scheme; if (!class_exists($this->protocol)) { $this->protocol = "\\Workerman\\Protocols\\$scheme"; if (!class_exists($this->protocol)) { throw new Exception("class \\Protocols\\$scheme not exist"); } } $local_socket = $this->transport . ":" . $address; } else { $this->transport = self::$_builtinTransports[$scheme]; } ~~~ 應用層協議檢測與初始化協議有關見另：Protocols協議 `$flags = $this->transport === 'udp' ? STREAM_SERVER_BIND : STREAM_SERVER_BIND | STREAM_SERVER_LISTEN;` 協議標識字段獲取 ~~~ if ($this->reusePort) { stream_context_set_option($this->_context, 'socket', 'so_reuseport', 1); } ~~~ 端口復用選項設置 ~~~ if ($this->transport === 'unix') { umask(0); list(, $address) = explode(':', $this->_socketName, 2); if (!is_file($address)) { register_shutdown_function(function () use ($address) { @unlink($address); }); } } $this->_mainSocket = stream_socket_server($local_socket, $errno, $errmsg, $flags, $this->_context); if (!$this->_mainSocket) { throw new Exception($errmsg); } ~~~ >[info] Uinx套接字協議首先檢查$address是否是文件然后創建unix套接字服務socket ~~~ if (function_exists('socket_import_stream') && $this->transport === 'tcp') { $socket = socket_import_stream($this->_mainSocket); @socket_set_option($socket, SOL_SOCKET, SO_KEEPALIVE, 1); @socket_set_option($socket, SOL_TCP, TCP_NODELAY, 1); } stream_set_blocking($this->_mainSocket, 0); ~~~ >[info] tcp協議處理 ~~~ if (self::$globalEvent) { 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')); } } ~~~ 注冊事件監聽器事件相關見另：Event事件 ### 7 signalHandler()[重點] `public static function signalHandler($signal) 主進程信號處理轉發` > $signal:待處理信號根據$signal，分別調用stopAll(),reload()或者 writeStatisticsToStatusFile() ### 8 stopAll()[重點] `public static function stopAll() 主進程stop信號處理` ~~~ self::$_status = self::STATUS_SHUTDOWN; ~~~ 設置當前狀態為關閉狀態 ~~~ if (self::$_masterPid === posix_getpid()) { self::log("Workerman[" . basename(self::$_startFile) . "] Stopping ..."); $worker_pid_array = self::getAllWorkerPids(); // Send stop signal to all child processes. foreach ($worker_pid_array as $worker_pid) { posix_kill($worker_pid, SIGINT); Timer::add(self::KILL_WORKER_TIMER_TIME, 'posix_kill', array($worker_pid, SIGKILL), false); } } ~~~ 主進程關閉處理記錄關閉信息獲取所有worker子進程id 發送關閉信號到所有worker子進程 ~~~ else { foreach (self::$_workers as $worker) { $worker->stop(); } exit(0); } ~~~ 子進程關閉處理直接關閉 ### 9 reload() `protected static function reload() 進程重載` `if (self::$_masterPid === posix_getpid()) {}` >[info] 主進程重載 ~~~ if (self::$_status !== self::STATUS_RELOADING && self::$_status !== self::STATUS_SHUTDOWN) { self::log("Workerman[" . basename(self::$_startFile) . "] reloading"); self::$_status = self::STATUS_RELOADING; } ~~~ 設置當前狀態為重載狀態 ~~~ $reloadable_pid_array = array(); foreach (self::$_pidMap as $worker_id => $worker_pid_array) { $worker = self::$_workers[$worker_id]; if ($worker->reloadable) { foreach ($worker_pid_array as $pid) { $reloadable_pid_array[$pid] = $pid; } } else { foreach ($worker_pid_array as $pid) { // Send reload signal to a worker process which reloadable is false. posix_kill($pid, SIGUSR1); } } } ~~~ 發送reload信號到子進程 `self::$_pidsToRestart = array_intersect(self::$_pidsToRestart, $reloadable_pid_array); ` 獲取所有等待重載進程id ~~~ if (empty(self::$_pidsToRestart)) { if (self::$_status !== self::STATUS_SHUTDOWN) { self::$_status = self::STATUS_RUNNING; } return; } ~~~ 檢測是否完全重載 ~~~ $one_worker_pid = current(self::$_pidsToRestart); posix_kill($one_worker_pid, SIGUSR1); ~~~ 發送reload信號到沒有完成重載的子進程 ~~~ $worker = current(self::$_workers); if ($worker->onWorkerReload) { try { call_user_func($worker->onWorkerReload, $worker); } catch (\Exception $e) { echo $e; exit(250); } } if ($worker->reloadable) { self::stopAll(); } ~~~ >[info] 子進程重載回調onWorkerReload函數不可重載則關閉所有 ### 10 protected static function writeStatisticsToStatusFile() `if (self::$_masterPid === posix_getpid()) {}` 主進程狀態信息保存子進程狀態新保存 ### 20 stop() `public function stop() 關閉worker子進程` 回調onWorkerStop 移除監聽器關閉socket描述符 ## 3 函數關系