Worker.php 45 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400
  1. <?php
  2. /**
  3. * This file is part of workerman.
  4. *
  5. * Licensed under The MIT License
  6. * For full copyright and license information, please see the MIT-LICENSE.txt
  7. * Redistributions of files must retain the above copyright notice.
  8. *
  9. * @author walkor<walkor@workerman.net>
  10. * @copyright walkor<walkor@workerman.net>
  11. * @link http://www.workerman.net/
  12. * @license http://www.opensource.org/licenses/mit-license.php MIT License
  13. */
  14. namespace Workerman;
  15. require_once __DIR__.'/Lib/Constants.php';
  16. use \Workerman\Events\Libevent;
  17. use \Workerman\Events\Select;
  18. use \Workerman\Events\EventInterface;
  19. use \Workerman\Connection\ConnectionInterface;
  20. use \Workerman\Connection\TcpConnection;
  21. use \Workerman\Connection\UdpConnection;
  22. use \Workerman\Lib\Timer;
  23. use \Exception;
  24. /**
  25. * Worker 类
  26. * 是一个容器,用于监听端口,维持客户端连接
  27. */
  28. class Worker
  29. {
  30. /**
  31. * 版本号
  32. * @var string
  33. */
  34. const VERSION = '3.2.0';
  35. /**
  36. * 状态 启动中
  37. * @var int
  38. */
  39. const STATUS_STARTING = 1;
  40. /**
  41. * 状态 运行中
  42. * @var int
  43. */
  44. const STATUS_RUNNING = 2;
  45. /**
  46. * 状态 停止
  47. * @var int
  48. */
  49. const STATUS_SHUTDOWN = 4;
  50. /**
  51. * 状态 平滑重启中
  52. * @var int
  53. */
  54. const STATUS_RELOADING = 8;
  55. /**
  56. * 给子进程发送重启命令 KILL_WORKER_TIMER_TIME 秒后
  57. * 如果对应进程仍然未重启则强行杀死
  58. * @var int
  59. */
  60. const KILL_WORKER_TIMER_TIME = 1;
  61. /**
  62. * 默认的backlog,即内核中用于存放未被进程认领(accept)的连接队列长度
  63. * @var int
  64. */
  65. const DEFAUL_BACKLOG = 1024;
  66. /**
  67. * udp最大包长
  68. * @var int
  69. */
  70. const MAX_UDP_PACKEG_SIZE = 65535;
  71. /**
  72. * worker的名称,用于在运行status命令时标记进程
  73. * @var string
  74. */
  75. public $name = 'none';
  76. /**
  77. * 设置当前worker实例的进程数
  78. * @var int
  79. */
  80. public $count = 1;
  81. /**
  82. * 设置当前worker进程的运行用户,启动时需要root超级权限
  83. * @var string
  84. */
  85. public $user = '';
  86. /**
  87. * 当前worker进程是否可以平滑重启
  88. * @var bool
  89. */
  90. public $reloadable = true;
  91. /**
  92. * 当worker进程启动时,如果设置了$onWorkerStart回调函数,则运行
  93. * 此钩子函数一般用于进程启动后初始化工作
  94. * @var callback
  95. */
  96. public $onWorkerStart = null;
  97. /**
  98. * 当有客户端连接时,如果设置了$onConnect回调函数,则运行
  99. * @var callback
  100. */
  101. public $onConnect = null;
  102. /**
  103. * 当客户端连接上发来数据时,如果设置了$onMessage回调,则运行
  104. * @var callback
  105. */
  106. public $onMessage = null;
  107. /**
  108. * 当客户端的连接关闭时,如果设置了$onClose回调,则运行
  109. * @var callback
  110. */
  111. public $onClose = null;
  112. /**
  113. * 当客户端的连接发生错误时,如果设置了$onError回调,则运行
  114. * 错误一般为客户端断开连接导致数据发送失败、服务端的发送缓冲区满导致发送失败等
  115. * 具体错误码及错误详情会以参数的形式传递给回调,参见手册
  116. * @var callback
  117. */
  118. public $onError = null;
  119. /**
  120. * 当连接的发送缓冲区满时,如果设置了$onBufferFull回调,则执行
  121. * @var callback
  122. */
  123. public $onBufferFull = null;
  124. /**
  125. * 当链接的发送缓冲区被清空时,如果设置了$onBufferDrain回调,则执行
  126. * @var callback
  127. */
  128. public $onBufferDrain = null;
  129. /**
  130. * 当前进程退出时(由于平滑重启或者服务停止导致),如果设置了此回调,则运行
  131. * @var callback
  132. */
  133. public $onWorkerStop = null;
  134. /**
  135. * 传输层协议
  136. * @var string
  137. */
  138. public $transport = 'tcp';
  139. /**
  140. * 所有的客户端连接
  141. * @var array
  142. */
  143. public $connections = array();
  144. /**
  145. * 应用层协议,由初始化worker时指定
  146. * 例如 new worker('http://0.0.0.0:8080');指定使用http协议
  147. * @var string
  148. */
  149. protected $_protocol = '';
  150. /**
  151. * 当前worker实例初始化目录位置,用于设置应用自动加载的根目录
  152. * @var string
  153. */
  154. protected $_appInitPath = '';
  155. /**
  156. * 是否以守护进程的方式运行。运行start时加上-d参数会自动以守护进程方式运行
  157. * 例如 php start.php start -d
  158. * @var bool
  159. */
  160. public static $daemonize = false;
  161. /**
  162. * 重定向标准输出,即将所有echo、var_dump等终端输出写到对应文件中
  163. * 注意 此参数只有在以守护进程方式运行时有效
  164. * @var string
  165. */
  166. public static $stdoutFile = '/dev/null';
  167. /**
  168. * pid文件的路径及名称
  169. * 例如 Worker::$pidFile = '/tmp/workerman.pid';
  170. * 注意 此属性一般不必手动设置,默认会放到php临时目录中
  171. * @var string
  172. */
  173. public static $pidFile = '';
  174. /**
  175. * 日志目录,默认在workerman根目录下,与Applications同级
  176. * 可以手动设置
  177. * 例如 Worker::$logFile = '/tmp/workerman.log';
  178. * @var mixed
  179. */
  180. public static $logFile = '';
  181. /**
  182. * 全局事件轮询库,用于监听所有资源的可读可写事件
  183. * @var Select/Libevent
  184. */
  185. public static $globalEvent = null;
  186. /**
  187. * 主进程pid
  188. * @var int
  189. */
  190. protected static $_masterPid = 0;
  191. /**
  192. * 监听的socket
  193. * @var resource
  194. */
  195. protected $_mainSocket = null;
  196. /**
  197. * socket名称,包括应用层协议+ip+端口号,在初始化worker时设置
  198. * 值类似 http://0.0.0.0:80
  199. * @var string
  200. */
  201. protected $_socketName = '';
  202. /**
  203. * socket的上下文,具体选项设置可以在初始化worker时传递
  204. * @var array
  205. */
  206. protected $_context = null;
  207. /**
  208. * 所有的worker实例
  209. * @var array
  210. */
  211. protected static $_workers = array();
  212. /**
  213. * 所有worker进程的pid
  214. * 格式为 [worker_id=>[pid=>pid, pid=>pid, ..], ..]
  215. * @var array
  216. */
  217. protected static $_pidMap = array();
  218. /**
  219. * 所有需要重启的进程pid
  220. * 格式为 [pid=>pid, pid=>pid]
  221. * @var array
  222. */
  223. protected static $_pidsToRestart = array();
  224. /**
  225. * 当前worker状态
  226. * @var int
  227. */
  228. protected static $_status = self::STATUS_STARTING;
  229. /**
  230. * 所有worke名称(name属性)中的最大长度,用于在运行 status 命令时格式化输出
  231. * @var int
  232. */
  233. protected static $_maxWorkerNameLength = 12;
  234. /**
  235. * 所有socket名称(_socketName属性)中的最大长度,用于在运行 status 命令时格式化输出
  236. * @var int
  237. */
  238. protected static $_maxSocketNameLength = 12;
  239. /**
  240. * 所有user名称(user属性)中的最大长度,用于在运行 status 命令时格式化输出
  241. * @var int
  242. */
  243. protected static $_maxUserNameLength = 12;
  244. /**
  245. * 运行 status 命令时用于保存结果的文件名
  246. * @var string
  247. */
  248. protected static $_statisticsFile = '';
  249. /**
  250. * 启动的全局入口文件
  251. * 例如 php start.php start ,则入口文件为start.php
  252. * @var string
  253. */
  254. protected static $_startFile = '';
  255. /**
  256. * 全局统计数据,用于在运行 status 命令时展示
  257. * 统计的内容包括 workerman启动的时间戳及每组worker进程的退出次数及退出状态码
  258. * @var array
  259. */
  260. protected static $_globalStatistics = array(
  261. 'start_timestamp' => 0,
  262. 'worker_exit_info' => array()
  263. );
  264. /**
  265. * 运行所有worker实例
  266. * @return void
  267. */
  268. public static function runAll()
  269. {
  270. // 初始化环境变量
  271. self::init();
  272. // 解析命令
  273. self::parseCommand();
  274. // 尝试以守护进程模式运行
  275. self::daemonize();
  276. // 初始化所有worker实例,主要是监听端口
  277. self::initWorkers();
  278. // 初始化所有信号处理函数
  279. self::installSignal();
  280. // 保存主进程pid
  281. self::saveMasterPid();
  282. // 创建子进程(worker进程)并运行
  283. self::forkWorkers();
  284. // 展示启动界面
  285. self::displayUI();
  286. // 尝试重定向标准输入输出
  287. self::resetStd();
  288. // 监控所有子进程(worker进程)
  289. self::monitorWorkers();
  290. }
  291. /**
  292. * 初始化一些环境变量
  293. * @return void
  294. */
  295. public static function init()
  296. {
  297. // 如果没设置$pidFile,则生成默认值
  298. if(empty(self::$pidFile))
  299. {
  300. $backtrace = debug_backtrace();
  301. self::$_startFile = $backtrace[count($backtrace)-1]['file'];
  302. self::$pidFile = sys_get_temp_dir()."/workerman.".str_replace('/', '_', self::$_startFile).".pid";
  303. }
  304. // 没有设置日志文件,则生成一个默认值
  305. if(empty(self::$logFile))
  306. {
  307. self::$logFile = __DIR__ . '/../workerman.log';
  308. }
  309. // 标记状态为启动中
  310. self::$_status = self::STATUS_STARTING;
  311. // 启动时间戳
  312. self::$_globalStatistics['start_timestamp'] = time();
  313. // 设置status文件位置
  314. self::$_statisticsFile = sys_get_temp_dir().'/workerman.status';
  315. // 尝试设置进程名称(需要php>=5.5或者安装了proctitle扩展)
  316. self::setProcessTitle('WorkerMan: master process start_file=' . self::$_startFile);
  317. // 初始化定时器
  318. Timer::init();
  319. }
  320. /**
  321. * 初始化所有的worker实例,主要工作为获得格式化所需数据及监听端口
  322. * @return void
  323. */
  324. protected static function initWorkers()
  325. {
  326. /** @var static $worker */
  327. foreach(self::$_workers as $worker)
  328. {
  329. // 没有设置worker名称,则使用none代替
  330. if(empty($worker->name))
  331. {
  332. $worker->name = 'none';
  333. }
  334. // 获得所有worker名称中最大长度
  335. $worker_name_length = strlen($worker->name);
  336. if(self::$_maxWorkerNameLength < $worker_name_length)
  337. {
  338. self::$_maxWorkerNameLength = $worker_name_length;
  339. }
  340. // 获得所有_socketName中最大长度
  341. $socket_name_length = strlen($worker->getSocketName());
  342. if(self::$_maxSocketNameLength < $socket_name_length)
  343. {
  344. self::$_maxSocketNameLength = $socket_name_length;
  345. }
  346. // 获得运行用户名的最大长度
  347. if(empty($worker->user) || posix_getuid() !== 0)
  348. {
  349. $worker->user = self::getCurrentUser();
  350. }
  351. $user_name_length = strlen($worker->user);
  352. if(self::$_maxUserNameLength < $user_name_length)
  353. {
  354. self::$_maxUserNameLength = $user_name_length;
  355. }
  356. // 监听端口
  357. $worker->listen();
  358. }
  359. }
  360. /**
  361. * 获得运行当前进程的用户名
  362. * @return string
  363. */
  364. protected static function getCurrentUser()
  365. {
  366. $user_info = posix_getpwuid(posix_getuid());
  367. return $user_info['name'];
  368. }
  369. /**
  370. * 展示启动界面
  371. * @return void
  372. */
  373. protected static function displayUI()
  374. {
  375. echo "\033[1A\n\033[K-----------------------\033[47;30m WORKERMAN \033[0m-----------------------------\n\033[0m";
  376. echo 'Workerman version:' , Worker::VERSION , " PHP version:",PHP_VERSION,"\n";
  377. echo "------------------------\033[47;30m WORKERS \033[0m-------------------------------\n";
  378. echo "\033[47;30muser\033[0m",str_pad('', self::$_maxUserNameLength+2-strlen('user')), "\033[47;30mworker\033[0m",str_pad('', self::$_maxWorkerNameLength+2-strlen('worker')), "\033[47;30mlisten\033[0m",str_pad('', self::$_maxSocketNameLength+2-strlen('listen')), "\033[47;30mprocesses\033[0m \033[47;30m","status\033[0m\n";
  379. /** @var static $worker */
  380. foreach(self::$_workers as $worker)
  381. {
  382. echo str_pad($worker->user, self::$_maxUserNameLength+2),str_pad($worker->name, self::$_maxWorkerNameLength+2),str_pad($worker->getSocketName(), self::$_maxSocketNameLength+2), str_pad(' '.$worker->count, 9), " \033[32;40m [OK] \033[0m\n";;
  383. }
  384. echo "----------------------------------------------------------------\n";
  385. if(self::$daemonize)
  386. {
  387. global $argv;
  388. $start_file = $argv[0];
  389. echo "Input \"php $start_file stop\" to quit. Start success.\n";
  390. }
  391. else
  392. {
  393. echo "Press Ctrl-C to quit. Start success.\n";
  394. }
  395. }
  396. /**
  397. * 解析运行命令
  398. * php yourfile.php start | stop | restart | reload | status
  399. * @return void
  400. */
  401. public static function parseCommand()
  402. {
  403. // 检查运行命令的参数
  404. global $argv;
  405. $start_file = $argv[0];
  406. if(!isset($argv[1]))
  407. {
  408. exit("Usage: php yourfile.php {start|stop|restart|reload|status}\n");
  409. }
  410. // 命令
  411. $command = trim($argv[1]);
  412. // 子命令,目前只支持-d
  413. $command2 = isset($argv[2]) ? $argv[2] : '';
  414. // 记录日志
  415. $mode = '';
  416. if($command === 'start')
  417. {
  418. if($command2 === '-d')
  419. {
  420. $mode = 'in DAEMON mode';
  421. }
  422. else
  423. {
  424. $mode = 'in DEBUG mode';
  425. }
  426. }
  427. self::log("Workerman[$start_file] $command $mode");
  428. // 检查主进程是否在运行
  429. $master_pid = @file_get_contents(self::$pidFile);
  430. $master_is_alive = $master_pid && @posix_kill($master_pid, 0);
  431. if($master_is_alive)
  432. {
  433. if($command === 'start')
  434. {
  435. self::log("Workerman[$start_file] already running");
  436. exit;
  437. }
  438. }
  439. elseif($command !== 'start' && $command !== 'restart')
  440. {
  441. self::log("Workerman[$start_file] not run");
  442. }
  443. // 根据命令做相应处理
  444. switch($command)
  445. {
  446. // 启动 workerman
  447. case 'start':
  448. if($command2 === '-d')
  449. {
  450. Worker::$daemonize = true;
  451. }
  452. break;
  453. // 显示 workerman 运行状态
  454. case 'status':
  455. // 尝试删除统计文件,避免脏数据
  456. if(is_file(self::$_statisticsFile))
  457. {
  458. @unlink(self::$_statisticsFile);
  459. }
  460. // 向主进程发送 SIGUSR2 信号 ,然后主进程会向所有子进程发送 SIGUSR2 信号
  461. // 所有进程收到 SIGUSR2 信号后会向 $_statisticsFile 写入自己的状态
  462. posix_kill($master_pid, SIGUSR2);
  463. // 睡眠100毫秒,等待子进程将自己的状态写入$_statisticsFile指定的文件
  464. usleep(100000);
  465. // 展示状态
  466. readfile(self::$_statisticsFile);
  467. exit(0);
  468. // 重启 workerman
  469. case 'restart':
  470. // 停止 workeran
  471. case 'stop':
  472. self::log("Workerman[$start_file] is stoping ...");
  473. // 想主进程发送SIGINT信号,主进程会向所有子进程发送SIGINT信号
  474. $master_pid && posix_kill($master_pid, SIGINT);
  475. // 如果 $timeout 秒后主进程没有退出则展示失败界面
  476. $timeout = 5;
  477. $start_time = time();
  478. while(1)
  479. {
  480. // 检查主进程是否存活
  481. $master_is_alive = $master_pid && posix_kill($master_pid, 0);
  482. if($master_is_alive)
  483. {
  484. // 检查是否超过$timeout时间
  485. if(time() - $start_time >= $timeout)
  486. {
  487. self::log("Workerman[$start_file] stop fail");
  488. exit;
  489. }
  490. usleep(10000);
  491. continue;
  492. }
  493. self::log("Workerman[$start_file] stop success");
  494. // 是restart命令
  495. if($command === 'stop')
  496. {
  497. exit(0);
  498. }
  499. // -d 说明是以守护进程的方式启动
  500. if($command2 === '-d')
  501. {
  502. Worker::$daemonize = true;
  503. }
  504. break;
  505. }
  506. break;
  507. // 平滑重启 workerman
  508. case 'reload':
  509. posix_kill($master_pid, SIGUSR1);
  510. self::log("Workerman[$start_file] reload");
  511. exit;
  512. // 未知命令
  513. default :
  514. exit("Usage: php yourfile.php {start|stop|restart|reload|status}\n");
  515. }
  516. }
  517. /**
  518. * 安装信号处理函数
  519. * @return void
  520. */
  521. protected static function installSignal()
  522. {
  523. // stop
  524. pcntl_signal(SIGINT, array('\Workerman\Worker', 'signalHandler'), false);
  525. // reload
  526. pcntl_signal(SIGUSR1, array('\Workerman\Worker', 'signalHandler'), false);
  527. // status
  528. pcntl_signal(SIGUSR2, array('\Workerman\Worker', 'signalHandler'), false);
  529. // ignore
  530. pcntl_signal(SIGPIPE, SIG_IGN, false);
  531. }
  532. /**
  533. * 为子进程重新安装信号处理函数,使用全局事件轮询监听信号
  534. * @return void
  535. */
  536. protected static function reinstallSignal()
  537. {
  538. // uninstall stop signal handler
  539. pcntl_signal(SIGINT, SIG_IGN, false);
  540. // uninstall reload signal handler
  541. pcntl_signal(SIGUSR1, SIG_IGN, false);
  542. // uninstall status signal handler
  543. pcntl_signal(SIGUSR2, SIG_IGN, false);
  544. // reinstall stop signal handler
  545. self::$globalEvent->add(SIGINT, EventInterface::EV_SIGNAL, array('\Workerman\Worker', 'signalHandler'));
  546. // uninstall reload signal handler
  547. self::$globalEvent->add(SIGUSR1, EventInterface::EV_SIGNAL,array('\Workerman\Worker', 'signalHandler'));
  548. // uninstall status signal handler
  549. self::$globalEvent->add(SIGUSR2, EventInterface::EV_SIGNAL, array('\Workerman\Worker', 'signalHandler'));
  550. }
  551. /**
  552. * 信号处理函数
  553. * @param int $signal
  554. */
  555. public static function signalHandler($signal)
  556. {
  557. switch($signal)
  558. {
  559. // stop
  560. case SIGINT:
  561. self::stopAll();
  562. break;
  563. // reload
  564. case SIGUSR1:
  565. self::$_pidsToRestart = self::getAllWorkerPids();
  566. self::reload();
  567. break;
  568. // show status
  569. case SIGUSR2:
  570. self::writeStatisticsToStatusFile();
  571. break;
  572. }
  573. }
  574. /**
  575. * 尝试以守护进程的方式运行
  576. * @throws Exception
  577. */
  578. protected static function daemonize()
  579. {
  580. if(!self::$daemonize)
  581. {
  582. return;
  583. }
  584. umask(0);
  585. $pid = pcntl_fork();
  586. if(-1 === $pid)
  587. {
  588. throw new Exception('fork fail');
  589. }
  590. elseif($pid > 0)
  591. {
  592. exit(0);
  593. }
  594. if(-1 === posix_setsid())
  595. {
  596. throw new Exception("setsid fail");
  597. }
  598. // fork again avoid SVR4 system regain the control of terminal
  599. $pid = pcntl_fork();
  600. if(-1 === $pid)
  601. {
  602. throw new Exception("fork fail");
  603. }
  604. elseif(0 !== $pid)
  605. {
  606. exit(0);
  607. }
  608. }
  609. /**
  610. * 重定向标准输入输出
  611. * @throws Exception
  612. */
  613. protected static function resetStd()
  614. {
  615. if(!self::$daemonize)
  616. {
  617. return;
  618. }
  619. global $STDOUT, $STDERR;
  620. $handle = fopen(self::$stdoutFile,"a");
  621. if($handle)
  622. {
  623. unset($handle);
  624. @fclose(STDOUT);
  625. @fclose(STDERR);
  626. $STDOUT = fopen(self::$stdoutFile,"a");
  627. $STDERR = fopen(self::$stdoutFile,"a");
  628. }
  629. else
  630. {
  631. throw new Exception('can not open stdoutFile ' . self::$stdoutFile);
  632. }
  633. }
  634. /**
  635. * 保存pid到文件中,方便运行命令时查找主进程pid
  636. * @throws Exception
  637. */
  638. protected static function saveMasterPid()
  639. {
  640. self::$_masterPid = posix_getpid();
  641. if(false === @file_put_contents(self::$pidFile, self::$_masterPid))
  642. {
  643. throw new Exception('can not save pid to ' . self::$pidFile);
  644. }
  645. }
  646. /**
  647. * 获得所有子进程的pid
  648. * @return array
  649. */
  650. protected static function getAllWorkerPids()
  651. {
  652. $pid_array = array();
  653. foreach(self::$_pidMap as $worker_pid_array)
  654. {
  655. foreach($worker_pid_array as $worker_pid)
  656. {
  657. $pid_array[$worker_pid] = $worker_pid;
  658. }
  659. }
  660. return $pid_array;
  661. }
  662. /**
  663. * 创建子进程
  664. * @return void
  665. */
  666. protected static function forkWorkers()
  667. {
  668. /** @var static $worker */
  669. foreach(self::$_workers as $worker)
  670. {
  671. // 启动过程中需要得到运行用户名的最大长度,在status时格式化展示
  672. if(self::$_status === self::STATUS_STARTING)
  673. {
  674. if(empty($worker->name))
  675. {
  676. $worker->name = $worker->getSocketName();
  677. }
  678. $worker_name_length = strlen($worker->name);
  679. if(self::$_maxWorkerNameLength < $worker_name_length)
  680. {
  681. self::$_maxWorkerNameLength = $worker_name_length;
  682. }
  683. }
  684. // 创建子进程
  685. while(count(self::$_pidMap[$worker->workerId]) < $worker->count)
  686. {
  687. static::forkOneWorker($worker);
  688. }
  689. }
  690. }
  691. /**
  692. * 创建一个子进程
  693. * @param Worker $worker
  694. * @throws Exception
  695. */
  696. protected static function forkOneWorker($worker)
  697. {
  698. $pid = pcntl_fork();
  699. // 主进程记录子进程pid
  700. if($pid > 0)
  701. {
  702. self::$_pidMap[$worker->workerId][$pid] = $pid;
  703. }
  704. // 子进程运行
  705. elseif(0 === $pid)
  706. {
  707. // 启动过程中尝试重定向标准输出
  708. if(self::$_status === self::STATUS_STARTING)
  709. {
  710. self::resetStd();
  711. }
  712. self::$_pidMap = array();
  713. self::$_workers = array($worker->workerId => $worker);
  714. Timer::delAll();
  715. self::setProcessTitle('WorkerMan: worker process ' . $worker->name . ' ' . $worker->getSocketName());
  716. self::setProcessUser($worker->user);
  717. $worker->run();
  718. exit(250);
  719. }
  720. else
  721. {
  722. throw new Exception("forkOneWorker fail");
  723. }
  724. }
  725. /**
  726. * 尝试设置运行当前进程的用户
  727. *
  728. * @param $user_name
  729. */
  730. protected static function setProcessUser($user_name)
  731. {
  732. if(empty($user_name) || posix_getuid() !== 0)
  733. {
  734. return;
  735. }
  736. $user_info = posix_getpwnam($user_name);
  737. if($user_info['uid'] != posix_getuid() || $user_info['gid'] != posix_getgid())
  738. {
  739. if(!posix_setgid($user_info['gid']) || !posix_setuid($user_info['uid']))
  740. {
  741. self::log( 'Notice : Can not run woker as '.$user_name." , You should be root\n", true);
  742. }
  743. }
  744. }
  745. /**
  746. * 设置当前进程的名称,在ps aux命令中有用
  747. * 注意 需要php>=5.5或者安装了protitle扩展
  748. * @param string $title
  749. * @return void
  750. */
  751. protected static function setProcessTitle($title)
  752. {
  753. // >=php 5.5
  754. if (function_exists('cli_set_process_title'))
  755. {
  756. @cli_set_process_title($title);
  757. }
  758. // 需要扩展
  759. elseif(extension_loaded('proctitle') && function_exists('setproctitle'))
  760. {
  761. @setproctitle($title);
  762. }
  763. }
  764. /**
  765. * 监控所有子进程的退出事件及退出码
  766. * @return void
  767. */
  768. protected static function monitorWorkers()
  769. {
  770. self::$_status = self::STATUS_RUNNING;
  771. while(1)
  772. {
  773. // 如果有信号到来,尝试触发信号处理函数
  774. pcntl_signal_dispatch();
  775. // 挂起进程,直到有子进程退出或者被信号打断
  776. $status = 0;
  777. $pid = pcntl_wait($status, WUNTRACED);
  778. // 如果有信号到来,尝试触发信号处理函数
  779. pcntl_signal_dispatch();
  780. // 有子进程退出
  781. if($pid > 0)
  782. {
  783. // 查找是哪个进程组的,然后再启动新的进程补上
  784. foreach(self::$_pidMap as $worker_id => $worker_pid_array)
  785. {
  786. if(isset($worker_pid_array[$pid]))
  787. {
  788. $worker = self::$_workers[$worker_id];
  789. // 检查退出状态
  790. if($status !== 0)
  791. {
  792. self::log("worker[".$worker->name.":$pid] exit with status $status");
  793. }
  794. // 统计,运行status命令时使用
  795. if(!isset(self::$_globalStatistics['worker_exit_info'][$worker_id][$status]))
  796. {
  797. self::$_globalStatistics['worker_exit_info'][$worker_id][$status] = 0;
  798. }
  799. self::$_globalStatistics['worker_exit_info'][$worker_id][$status]++;
  800. // 清除子进程信息
  801. unset(self::$_pidMap[$worker_id][$pid]);
  802. break;
  803. }
  804. }
  805. // 如果不是关闭状态,则补充新的进程
  806. if(self::$_status !== self::STATUS_SHUTDOWN)
  807. {
  808. self::forkWorkers();
  809. // 如果该进程是因为运行reload命令退出,则继续执行reload流程
  810. if(isset(self::$_pidsToRestart[$pid]))
  811. {
  812. unset(self::$_pidsToRestart[$pid]);
  813. self::reload();
  814. }
  815. }
  816. else
  817. {
  818. // 如果是关闭状态,并且所有进程退出完毕,则主进程退出
  819. if(!self::getAllWorkerPids())
  820. {
  821. self::exitAndClearAll();
  822. }
  823. }
  824. }
  825. else
  826. {
  827. // 如果是关闭状态,并且所有进程退出完毕,则主进程退出
  828. if(self::$_status === self::STATUS_SHUTDOWN && !self::getAllWorkerPids())
  829. {
  830. self::exitAndClearAll();
  831. }
  832. }
  833. }
  834. }
  835. /**
  836. * 退出当前进程
  837. * @return void
  838. */
  839. protected static function exitAndClearAll()
  840. {
  841. @unlink(self::$pidFile);
  842. self::log("Workerman[".basename(self::$_startFile)."] has been stopped");
  843. exit(0);
  844. }
  845. /**
  846. * 执行平滑重启流程
  847. * @return void
  848. */
  849. protected static function reload()
  850. {
  851. // 主进程部分
  852. if(self::$_masterPid === posix_getpid())
  853. {
  854. // 设置为平滑重启状态
  855. if(self::$_status !== self::STATUS_RELOADING && self::$_status !== self::STATUS_SHUTDOWN)
  856. {
  857. self::log("Workerman[".basename(self::$_startFile)."] reloading");
  858. self::$_status = self::STATUS_RELOADING;
  859. }
  860. // 如果有worker设置了reloadable=false,则过滤掉
  861. $reloadable_pid_array = array();
  862. foreach(self::$_pidMap as $worker_id =>$worker_pid_array)
  863. {
  864. $worker = self::$_workers[$worker_id];
  865. if($worker->reloadable)
  866. {
  867. foreach($worker_pid_array as $pid)
  868. {
  869. $reloadable_pid_array[$pid] = $pid;
  870. }
  871. }
  872. }
  873. // 得到所有可以重启的进程
  874. self::$_pidsToRestart = array_intersect(self::$_pidsToRestart , $reloadable_pid_array);
  875. // 平滑重启完毕
  876. if(empty(self::$_pidsToRestart))
  877. {
  878. if(self::$_status !== self::STATUS_SHUTDOWN)
  879. {
  880. self::$_status = self::STATUS_RUNNING;
  881. }
  882. return;
  883. }
  884. // 继续执行平滑重启流程
  885. $one_worker_pid = current(self::$_pidsToRestart );
  886. // 给子进程发送平滑重启信号
  887. posix_kill($one_worker_pid, SIGUSR1);
  888. // 定时器,如果子进程在KILL_WORKER_TIMER_TIME秒后没有退出,则强行杀死
  889. Timer::add(self::KILL_WORKER_TIMER_TIME, 'posix_kill', array($one_worker_pid, SIGKILL), false);
  890. }
  891. // 子进程部分
  892. else
  893. {
  894. // 如果当前worker的reloadable属性为真,则执行退出
  895. $worker = current(self::$_workers);
  896. if($worker->reloadable)
  897. {
  898. self::stopAll();
  899. }
  900. }
  901. }
  902. /**
  903. * 执行关闭流程
  904. * @return void
  905. */
  906. public static function stopAll()
  907. {
  908. self::$_status = self::STATUS_SHUTDOWN;
  909. // 主进程部分
  910. if(self::$_masterPid === posix_getpid())
  911. {
  912. self::log("Workerman[".basename(self::$_startFile)."] Stopping ...");
  913. $worker_pid_array = self::getAllWorkerPids();
  914. // 向所有子进程发送SIGINT信号,表明关闭服务
  915. foreach($worker_pid_array as $worker_pid)
  916. {
  917. posix_kill($worker_pid, SIGINT);
  918. Timer::add(self::KILL_WORKER_TIMER_TIME, 'posix_kill', array($worker_pid, SIGKILL),false);
  919. }
  920. }
  921. // 子进程部分
  922. else
  923. {
  924. // 执行stop逻辑
  925. /** @var static $worker */
  926. foreach(self::$_workers as $worker)
  927. {
  928. $worker->stop();
  929. }
  930. exit(0);
  931. }
  932. }
  933. /**
  934. * 将当前进程的统计信息写入到统计文件
  935. * @return void
  936. */
  937. protected static function writeStatisticsToStatusFile()
  938. {
  939. // 主进程部分
  940. if(self::$_masterPid === posix_getpid())
  941. {
  942. $loadavg = sys_getloadavg();
  943. file_put_contents(self::$_statisticsFile, "---------------------------------------GLOBAL STATUS--------------------------------------------\n");
  944. file_put_contents(self::$_statisticsFile, 'Workerman version:' . Worker::VERSION . " PHP version:".PHP_VERSION."\n", FILE_APPEND);
  945. file_put_contents(self::$_statisticsFile, 'start time:'. date('Y-m-d H:i:s', self::$_globalStatistics['start_timestamp']).' run ' . floor((time()-self::$_globalStatistics['start_timestamp'])/(24*60*60)). ' days ' . floor(((time()-self::$_globalStatistics['start_timestamp'])%(24*60*60))/(60*60)) . " hours \n", FILE_APPEND);
  946. file_put_contents(self::$_statisticsFile, 'load average: ' . implode(", ", $loadavg) . "\n", FILE_APPEND);
  947. file_put_contents(self::$_statisticsFile, count(self::$_pidMap) . ' workers ' . count(self::getAllWorkerPids())." processes\n", FILE_APPEND);
  948. file_put_contents(self::$_statisticsFile, str_pad('worker_name', self::$_maxWorkerNameLength) . " exit_status exit_count\n", FILE_APPEND);
  949. foreach(self::$_pidMap as $worker_id =>$worker_pid_array)
  950. {
  951. $worker = self::$_workers[$worker_id];
  952. if(isset(self::$_globalStatistics['worker_exit_info'][$worker_id]))
  953. {
  954. foreach(self::$_globalStatistics['worker_exit_info'][$worker_id] as $worker_exit_status=>$worker_exit_count)
  955. {
  956. file_put_contents(self::$_statisticsFile, str_pad($worker->name, self::$_maxWorkerNameLength) . " " . str_pad($worker_exit_status, 16). " $worker_exit_count\n", FILE_APPEND);
  957. }
  958. }
  959. else
  960. {
  961. file_put_contents(self::$_statisticsFile, str_pad($worker->name, self::$_maxWorkerNameLength) . " " . str_pad(0, 16). " 0\n", FILE_APPEND);
  962. }
  963. }
  964. file_put_contents(self::$_statisticsFile, "---------------------------------------PROCESS STATUS-------------------------------------------\n", FILE_APPEND);
  965. file_put_contents(self::$_statisticsFile, "pid\tmemory ".str_pad('listening', self::$_maxSocketNameLength)." ".str_pad('worker_name', self::$_maxWorkerNameLength)." connections ".str_pad('total_request', 13)." ".str_pad('send_fail', 9)." ".str_pad('throw_exception', 15)."\n", FILE_APPEND);
  966. chmod(self::$_statisticsFile, 0722);
  967. foreach(self::getAllWorkerPids() as $worker_pid)
  968. {
  969. posix_kill($worker_pid, SIGUSR2);
  970. }
  971. return;
  972. }
  973. // 子进程部分
  974. $worker = current(self::$_workers);
  975. $wrker_status_str = posix_getpid()."\t".str_pad(round(memory_get_usage(true)/(1024*1024),2)."M", 7)." " .str_pad($worker->getSocketName(), self::$_maxSocketNameLength) ." ".str_pad(($worker->name === $worker->getSocketName() ? 'none' : $worker->name), self::$_maxWorkerNameLength)." ";
  976. $wrker_status_str .= str_pad(ConnectionInterface::$statistics['connection_count'], 11)." ".str_pad(ConnectionInterface::$statistics['total_request'], 14)." ".str_pad(ConnectionInterface::$statistics['send_fail'],9)." ".str_pad(ConnectionInterface::$statistics['throw_exception'],15)."\n";
  977. file_put_contents(self::$_statisticsFile, $wrker_status_str, FILE_APPEND);
  978. }
  979. /**
  980. * 检查错误
  981. * @return void
  982. */
  983. public static function checkErrors()
  984. {
  985. if(self::STATUS_SHUTDOWN != self::$_status)
  986. {
  987. $error_msg = "WORKER EXIT UNEXPECTED ";
  988. $errors = error_get_last();
  989. if($errors && ($errors['type'] === E_ERROR ||
  990. $errors['type'] === E_PARSE ||
  991. $errors['type'] === E_CORE_ERROR ||
  992. $errors['type'] === E_COMPILE_ERROR ||
  993. $errors['type'] === E_RECOVERABLE_ERROR ))
  994. {
  995. $error_msg .= self::getErrorType($errors['type']) . " {$errors['message']} in {$errors['file']} on line {$errors['line']}";
  996. }
  997. self::log($error_msg);
  998. }
  999. }
  1000. /**
  1001. * 获取错误类型对应的意义
  1002. * @param integer $type
  1003. * @return string
  1004. */
  1005. protected static function getErrorType($type)
  1006. {
  1007. switch($type)
  1008. {
  1009. case E_ERROR: // 1 //
  1010. return 'E_ERROR';
  1011. case E_WARNING: // 2 //
  1012. return 'E_WARNING';
  1013. case E_PARSE: // 4 //
  1014. return 'E_PARSE';
  1015. case E_NOTICE: // 8 //
  1016. return 'E_NOTICE';
  1017. case E_CORE_ERROR: // 16 //
  1018. return 'E_CORE_ERROR';
  1019. case E_CORE_WARNING: // 32 //
  1020. return 'E_CORE_WARNING';
  1021. case E_COMPILE_ERROR: // 64 //
  1022. return 'E_COMPILE_ERROR';
  1023. case E_COMPILE_WARNING: // 128 //
  1024. return 'E_COMPILE_WARNING';
  1025. case E_USER_ERROR: // 256 //
  1026. return 'E_USER_ERROR';
  1027. case E_USER_WARNING: // 512 //
  1028. return 'E_USER_WARNING';
  1029. case E_USER_NOTICE: // 1024 //
  1030. return 'E_USER_NOTICE';
  1031. case E_STRICT: // 2048 //
  1032. return 'E_STRICT';
  1033. case E_RECOVERABLE_ERROR: // 4096 //
  1034. return 'E_RECOVERABLE_ERROR';
  1035. case E_DEPRECATED: // 8192 //
  1036. return 'E_DEPRECATED';
  1037. case E_USER_DEPRECATED: // 16384 //
  1038. return 'E_USER_DEPRECATED';
  1039. }
  1040. return "";
  1041. }
  1042. /**
  1043. * 记录日志
  1044. * @param string $msg
  1045. * @return void
  1046. */
  1047. protected static function log($msg)
  1048. {
  1049. $msg = $msg."\n";
  1050. if(!self::$daemonize)
  1051. {
  1052. echo $msg;
  1053. }
  1054. file_put_contents(self::$logFile, date('Y-m-d H:i:s') . " " . $msg, FILE_APPEND | LOCK_EX);
  1055. }
  1056. /**
  1057. * worker构造函数
  1058. *
  1059. * @param string $socket_name
  1060. * @param array $context_option
  1061. */
  1062. public function __construct($socket_name = '', $context_option = array())
  1063. {
  1064. // 保存worker实例
  1065. $this->workerId = spl_object_hash($this);
  1066. self::$_workers[$this->workerId] = $this;
  1067. self::$_pidMap[$this->workerId] = array();
  1068. // 获得实例化文件路径,用于自动加载设置根目录
  1069. $backrace = debug_backtrace();
  1070. $this->_appInitPath = dirname($backrace[0]['file']);
  1071. // 设置socket上下文
  1072. if($socket_name)
  1073. {
  1074. $this->_socketName = $socket_name;
  1075. if(!isset($context_option['socket']['backlog']))
  1076. {
  1077. $context_option['socket']['backlog'] = self::DEFAUL_BACKLOG;
  1078. }
  1079. $this->_context = stream_context_create($context_option);
  1080. }
  1081. // 设置一个空的onMessage,当onMessage未设置时用来消费socket数据
  1082. $this->onMessage = function(){};
  1083. }
  1084. /**
  1085. * 监听端口
  1086. * @throws Exception
  1087. */
  1088. public function listen()
  1089. {
  1090. // 设置自动加载根目录
  1091. Autoloader::setRootPath($this->_appInitPath);
  1092. if(!$this->_socketName)
  1093. {
  1094. return;
  1095. }
  1096. // 获得应用层通讯协议以及监听的地址
  1097. list($scheme, $address) = explode(':', $this->_socketName, 2);
  1098. // 如果有指定应用层协议,则检查对应的协议类是否存在
  1099. if($scheme != 'tcp' && $scheme != 'udp')
  1100. {
  1101. $scheme = ucfirst($scheme);
  1102. $this->_protocol = '\\Protocols\\'.$scheme;
  1103. if(!class_exists($this->_protocol))
  1104. {
  1105. $this->_protocol = "\\Workerman\\Protocols\\$scheme";
  1106. if(!class_exists($this->_protocol))
  1107. {
  1108. throw new Exception("class \\Protocols\\$scheme not exist");
  1109. }
  1110. }
  1111. }
  1112. elseif($scheme === 'udp')
  1113. {
  1114. $this->transport = 'udp';
  1115. }
  1116. // flag
  1117. $flags = $this->transport === 'udp' ? STREAM_SERVER_BIND : STREAM_SERVER_BIND | STREAM_SERVER_LISTEN;
  1118. $errno = 0;
  1119. $errmsg = '';
  1120. $this->_mainSocket = stream_socket_server($this->transport.":".$address, $errno, $errmsg, $flags, $this->_context);
  1121. if(!$this->_mainSocket)
  1122. {
  1123. throw new Exception($errmsg);
  1124. }
  1125. // 尝试打开tcp的keepalive,关闭TCP Nagle算法
  1126. if(function_exists('socket_import_stream'))
  1127. {
  1128. $socket = socket_import_stream($this->_mainSocket );
  1129. @socket_set_option($socket, SOL_SOCKET, SO_KEEPALIVE, 1);
  1130. @socket_set_option($socket, SOL_SOCKET, TCP_NODELAY, 1);
  1131. }
  1132. // 设置非阻塞
  1133. stream_set_blocking($this->_mainSocket, 0);
  1134. // 放到全局事件轮询中监听_mainSocket可读事件(客户端连接事件)
  1135. if(self::$globalEvent)
  1136. {
  1137. if($this->transport !== 'udp')
  1138. {
  1139. self::$globalEvent->add($this->_mainSocket, EventInterface::EV_READ, array($this, 'acceptConnection'));
  1140. }
  1141. else
  1142. {
  1143. self::$globalEvent->add($this->_mainSocket, EventInterface::EV_READ, array($this, 'acceptUdpConnection'));
  1144. }
  1145. }
  1146. }
  1147. /**
  1148. * 获得 socket name
  1149. * @return string
  1150. */
  1151. public function getSocketName()
  1152. {
  1153. return $this->_socketName ? $this->_socketName : 'none';
  1154. }
  1155. /**
  1156. * 运行worker实例
  1157. */
  1158. public function run()
  1159. {
  1160. //更新 Worker 状态
  1161. self::$_status = self::STATUS_RUNNING;
  1162. // 注册进程退出回调,用来检查是否有错误
  1163. register_shutdown_function(array("\\Workerman\\Worker", 'checkErrors'));
  1164. // 设置自动加载根目录
  1165. Autoloader::setRootPath($this->_appInitPath);
  1166. // 如果没有全局事件轮询,则创建一个
  1167. if(!self::$globalEvent)
  1168. {
  1169. if(extension_loaded('libevent'))
  1170. {
  1171. self::$globalEvent = new Libevent();
  1172. }
  1173. else
  1174. {
  1175. self::$globalEvent = new Select();
  1176. }
  1177. // 监听_mainSocket上的可读事件(客户端连接事件)
  1178. if($this->_socketName)
  1179. {
  1180. if($this->transport !== 'udp')
  1181. {
  1182. self::$globalEvent->add($this->_mainSocket, EventInterface::EV_READ, array($this, 'acceptConnection'));
  1183. }
  1184. else
  1185. {
  1186. self::$globalEvent->add($this->_mainSocket, EventInterface::EV_READ, array($this, 'acceptUdpConnection'));
  1187. }
  1188. }
  1189. }
  1190. // 重新安装事件处理函数,使用全局事件轮询监听信号事件
  1191. self::reinstallSignal();
  1192. // 用全局事件轮询初始化定时器
  1193. Timer::init(self::$globalEvent);
  1194. // 如果有设置进程启动回调,则执行
  1195. if($this->onWorkerStart)
  1196. {
  1197. call_user_func($this->onWorkerStart, $this);
  1198. }
  1199. // 子进程主循环
  1200. self::$globalEvent->loop();
  1201. }
  1202. /**
  1203. * 停止当前worker实例
  1204. * @return void
  1205. */
  1206. public function stop()
  1207. {
  1208. // 如果有设置进程终止回调,则执行
  1209. if($this->onWorkerStop)
  1210. {
  1211. call_user_func($this->onWorkerStop, $this);
  1212. }
  1213. // 删除相关监听事件,关闭_mainSocket
  1214. self::$globalEvent->del($this->_mainSocket, EventInterface::EV_READ);
  1215. @fclose($this->_mainSocket);
  1216. }
  1217. /**
  1218. * 接收一个客户端连接
  1219. * @param resource $socket
  1220. * @return void
  1221. */
  1222. public function acceptConnection($socket)
  1223. {
  1224. // 获得客户端连接
  1225. $new_socket = @stream_socket_accept($socket, 0);
  1226. // 惊群现象,忽略
  1227. if(false === $new_socket)
  1228. {
  1229. return;
  1230. }
  1231. // 初始化连接对象
  1232. $connection = new TcpConnection($new_socket);
  1233. $this->connections[$connection->id] = $connection;
  1234. $connection->worker = $this;
  1235. $connection->protocol = $this->_protocol;
  1236. $connection->onMessage = $this->onMessage;
  1237. $connection->onClose = $this->onClose;
  1238. $connection->onError = $this->onError;
  1239. $connection->onBufferDrain = $this->onBufferDrain;
  1240. $connection->onBufferFull = $this->onBufferFull;
  1241. // 如果有设置连接回调,则执行
  1242. if($this->onConnect)
  1243. {
  1244. try
  1245. {
  1246. call_user_func($this->onConnect, $connection);
  1247. }
  1248. catch(Exception $e)
  1249. {
  1250. ConnectionInterface::$statistics['throw_exception']++;
  1251. self::log($e);
  1252. }
  1253. }
  1254. }
  1255. /**
  1256. * 处理udp连接(udp其实是无连接的,这里为保证和tcp连接接口一致)
  1257. *
  1258. * @param resource $socket
  1259. * @return bool
  1260. */
  1261. public function acceptUdpConnection($socket)
  1262. {
  1263. $recv_buffer = stream_socket_recvfrom($socket , self::MAX_UDP_PACKEG_SIZE, 0, $remote_address);
  1264. if(false === $recv_buffer || empty($remote_address))
  1265. {
  1266. return false;
  1267. }
  1268. // 模拟一个连接对象
  1269. $connection = new UdpConnection($socket, $remote_address);
  1270. if($this->onMessage)
  1271. {
  1272. if($this->_protocol)
  1273. {
  1274. /** @var \Workerman\Protocols\ProtocolInterface $parser */
  1275. $parser = $this->_protocol;
  1276. $recv_buffer = $parser::decode($recv_buffer, $connection);
  1277. }
  1278. ConnectionInterface::$statistics['total_request']++;
  1279. try
  1280. {
  1281. call_user_func($this->onMessage, $connection, $recv_buffer);
  1282. }
  1283. catch(Exception $e)
  1284. {
  1285. ConnectionInterface::$statistics['throw_exception']++;
  1286. }
  1287. }
  1288. }
  1289. }