Worker.php 45 KB

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