Gateway.php 29 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877
  1. <?php
  2. /**
  3. *
  4. * 暴露给客户端的连接网关 只负责网络io
  5. * 1、监听客户端连接
  6. * 2、监听后端回应并转发回应给前端
  7. *
  8. * @author walkor <walkor@workerman.net>
  9. *
  10. */
  11. require_once __DIR__ . '/../Lib/Autoloader.php';
  12. use \Protocols\GatewayProtocol;
  13. use \Lib\Store;
  14. use \Lib\StatisticClient;
  15. class Gateway extends Man\Core\SocketWorker
  16. {
  17. /**
  18. * 内部通信socket udp
  19. * @var resouce
  20. */
  21. protected $innerMainSocketUdp = null;
  22. /**
  23. * 内部通信socket tcp
  24. * @var resouce
  25. */
  26. protected $innerMainSocketTcp = null;
  27. /**
  28. * 内网ip
  29. * @var string
  30. */
  31. protected $lanIp = '127.0.0.1';
  32. /**
  33. * 内部通信端口
  34. * @var int
  35. */
  36. protected $lanPort = 0;
  37. /**
  38. * client_id到连接的映射
  39. * @var array
  40. */
  41. protected $clientConnMap = array();
  42. /**
  43. * 连接到client_id的映射
  44. * @var array
  45. */
  46. protected $connClientMap = array();
  47. /**
  48. * 客户端链接和客户端远程地址映射
  49. * @var array
  50. */
  51. protected $connRemoteAddressMap = array();
  52. /**
  53. * client_id到session的映射
  54. * @var array
  55. */
  56. protected $connSessionMap = array();
  57. /**
  58. * 与worker的连接
  59. * [fd=>fd, $fd=>fd, ..]
  60. * @var array
  61. */
  62. protected $workerConnections = array();
  63. /**
  64. * 客户端心跳信息
  65. * [client_id=>not_response_count, client_id=>not_response_count, ..]
  66. */
  67. protected $pingInfo = array();
  68. /**
  69. * gateway 发送心跳时间间隔 单位:秒 ,0表示不发送心跳,在配置中设置
  70. * @var integer
  71. */
  72. protected $pingInterval = 0;
  73. /**
  74. * 心跳数据
  75. * 可以是二进制数据(二进制数据保存在文件中,在配置中设置ping数据文件路径 如 ping_data=/yourpath/ping.bin)
  76. * ping数据应该是客户端能够识别的数据格式,客户端必须回复心跳,不然链接会断开
  77. * @var string
  78. */
  79. protected $pingData = '';
  80. /**
  81. * 客户端连续$pingNotResponseLimit次不回应心跳则断开链接
  82. */
  83. protected $pingNotResponseLimit = 0;
  84. /**
  85. * 命令字,统计用到
  86. * @var array
  87. */
  88. protected static $interfaceMap = array(
  89. GatewayProtocol::CMD_SEND_TO_ONE => 'CMD_SEND_TO_ONE',
  90. GatewayProtocol::CMD_SEND_TO_ALL => 'CMD_SEND_TO_ALL',
  91. GatewayProtocol::CMD_KICK => 'CMD_KICK',
  92. GatewayProtocol::CMD_UPDATE_SESSION => 'CMD_UPDATE_SESSION',
  93. GatewayProtocol::CMD_GET_ONLINE_STATUS => 'CMD_GET_ONLINE_STATUS',
  94. GatewayProtocol::CMD_IS_ONLINE => 'CMD_IS_ONLINE',
  95. );
  96. /**
  97. * 由于网络延迟或者socket缓冲区大小的限制,客户端发来的数据可能不会都全部到达,需要根据协议判断数据是否完整
  98. * @see Man\Core.SocketWorker::dealInput()
  99. */
  100. public function dealInput($recv_buffer)
  101. {
  102. // 处理粘包
  103. return Event::onGatewayMessage($recv_buffer);
  104. }
  105. /**
  106. * 用户客户端发来消息时处理
  107. * @see Man\Core.SocketWorker::dealProcess()
  108. */
  109. public function dealProcess($recv_buffer)
  110. {
  111. // 客户端发来任何一个完整的包都视为回应了服务端发的心跳
  112. if(!empty($this->pingInfo[$this->connClientMap[$this->currentDealFd]]))
  113. {
  114. $this->pingInfo[$this->connClientMap[$this->currentDealFd]] = 0;
  115. }
  116. // 统计打点
  117. StatisticClient::tick(__CLASS__, 'CMD_ON_MESSAGE');
  118. // 触发ON_MESSAGE
  119. if(false === $this->sendToWorker(GatewayProtocol::CMD_ON_MESSAGE, $this->currentDealFd, $recv_buffer))
  120. {
  121. return StatisticClient::report(__CLASS__, 'CMD_ON_MESSAGE', 0, 132, __CLASS__.'::dealProcess()->sendToWorker() fail');
  122. }
  123. StatisticClient::report(__CLASS__, 'CMD_ON_MESSAGE', 1, 0, '');
  124. }
  125. /**
  126. * 进程启动
  127. */
  128. public function start()
  129. {
  130. // 安装信号处理函数
  131. $this->installSignal();
  132. // 添加accept事件
  133. $ret = $this->event->add($this->mainSocket, Man\Core\Events\BaseEvent::EV_READ, array($this, 'accept'));
  134. // 创建内部通信套接字,用于与BusinessWorker通讯
  135. $start_port = Man\Core\Lib\Config::get($this->workerName.'.lan_port_start');
  136. // 计算本进程监听的ip端口
  137. if(strpos(\Man\Core\Master::VERSION, 'mt'))
  138. {
  139. $this->lanPort = $start_port + \Thread::getCurrentThreadId()%100;
  140. }
  141. else
  142. {
  143. $this->lanPort = $start_port - posix_getppid() + posix_getpid();
  144. }
  145. // 如果端口不在合法范围
  146. if($this->lanPort<0 || $this->lanPort >=65535)
  147. {
  148. $this->lanPort = rand($start_port, 65535);
  149. }
  150. // 如果
  151. $this->lanIp = Man\Core\Lib\Config::get($this->workerName.'.lan_ip');
  152. if(!$this->lanIp)
  153. {
  154. $this->notice($this->workerName.'.lan_ip not set');
  155. $this->lanIp = '127.0.0.1';
  156. }
  157. $error_no_udp = $error_no_tcp = 0;
  158. $error_msg_udp = $error_msg_tcp = '';
  159. // 执行监听
  160. $this->innerMainSocketUdp = stream_socket_server("udp://".$this->lanIp.':'.$this->lanPort, $error_no_udp, $error_msg_udp, STREAM_SERVER_BIND);
  161. $this->innerMainSocketTcp = stream_socket_server("tcp://".$this->lanIp.':'.$this->lanPort, $error_no_tcp, $error_msg_tcp, STREAM_SERVER_BIND | STREAM_SERVER_LISTEN);
  162. // 出错,退出,下次会换个端口
  163. if(!$this->innerMainSocketUdp || !$this->innerMainSocketTcp)
  164. {
  165. $this->notice('create innerMainSocket udp or tcp fail and exit '.$error_msg_udp.$error_msg_tcp);
  166. $this->stop();
  167. }
  168. else
  169. {
  170. stream_set_blocking($this->innerMainSocketUdp , 0);
  171. stream_set_blocking($this->innerMainSocketTcp , 0);
  172. }
  173. // 注册套接字
  174. if(!$this->registerAddress($this->lanIp.':'.$this->lanPort))
  175. {
  176. $this->notice('registerAddress fail and exit');
  177. $this->stop();
  178. }
  179. // 添加读udp/tcp事件
  180. $this->event->add($this->innerMainSocketUdp, Man\Core\Events\BaseEvent::EV_READ, array($this, 'recvInnerUdp'));
  181. $this->event->add($this->innerMainSocketTcp, Man\Core\Events\BaseEvent::EV_READ, array($this, 'acceptInnerTcp'));
  182. // 初始化心跳包时间间隔
  183. $ping_interval = \Man\Core\Lib\Config::get($this->workerName.'.ping_interval');
  184. if((int)$ping_interval > 0)
  185. {
  186. $this->pingInterval = (int)$ping_interval;
  187. }
  188. // 获取心跳包数据
  189. $ping_data_or_path = \Man\Core\Lib\Config::get($this->workerName.'.ping_data');
  190. if(is_file($ping_data_or_path))
  191. {
  192. $this->pingData = file_get_contents($ping_data_or_path);
  193. }
  194. else
  195. {
  196. $this->pingData = $ping_data_or_path;
  197. }
  198. // 不返回心跳回应(客户端发来的任何数据都算是回应)的限定值
  199. $this->pingNotResponseLimit = (int)\Man\Core\Lib\Config::get($this->workerName.'.ping_not_response_limit');
  200. // 设置定时任务,发送心跳
  201. if($this->pingInterval > 0)
  202. {
  203. \Man\Core\Lib\Task::init($this->event);
  204. \Man\Core\Lib\Task::add($this->pingInterval, array($this, 'ping'));
  205. }
  206. // 主体循环,整个子进程会阻塞在这个函数上
  207. $ret = $this->event->loop();
  208. // 下面正常不会执行到
  209. $this->notice('worker loop exit');
  210. // 执行到就退出
  211. exit(0);
  212. }
  213. /**
  214. * 接受一个链接
  215. * @param resource $socket
  216. * @param $null_one $flag
  217. * @param $null_two $base
  218. * @return void
  219. */
  220. public function accept($socket, $null_one = null, $null_two = null)
  221. {
  222. // 获得一个连接
  223. $new_connection = @stream_socket_accept($socket, 0);
  224. // 可能是惊群效应
  225. if(false === $new_connection)
  226. {
  227. $this->statusInfo['thunder_herd']++;
  228. return false;
  229. }
  230. // 连接的fd序号
  231. $fd = (int) $new_connection;
  232. $this->connections[$fd] = $new_connection;
  233. $this->recvBuffers[$fd] = array('buf'=>'', 'remain_len'=>$this->prereadLength);
  234. $this->connSessionMap[$fd] = '';
  235. // 非阻塞
  236. stream_set_blocking($this->connections[$fd], 0);
  237. $this->event->add($this->connections[$fd], Man\Core\Events\BaseEvent::EV_READ , array($this, 'dealInputBase'), $fd);
  238. // 全局唯一client_id
  239. $global_client_id = $this->createGlobalClientId();
  240. $this->clientConnMap[$global_client_id] = $fd;
  241. $this->connClientMap[$fd] = $global_client_id;
  242. $address = array('local_ip'=>$this->lanIp, 'local_port'=>$this->lanPort, 'socket_id'=>$fd);
  243. // 保存客户端内部通讯地址
  244. $this->storeClientAddress($global_client_id, $address);
  245. // 客户端保存 ip:port
  246. $address= $this->getRemoteAddress($fd);
  247. if($address)
  248. {
  249. list($client_ip, $client_port) = explode(':', $address, 2);
  250. }
  251. else
  252. {
  253. $client_ip = 0;
  254. $client_port = 0;
  255. }
  256. $this->connRemoteAddressMap[$fd] = array('ip'=>$client_ip, 'port'=>$client_port);
  257. // 触发GatewayOnConnection事件
  258. if(method_exists('Event','onGatewayConnect'))
  259. {
  260. // 统计打点
  261. StatisticClient::tick(__CLASS__, 'CMD_ON_GATEWAY_CONNECTION');
  262. if(false === $this->sendToWorker(GatewayProtocol::CMD_ON_GATEWAY_CONNECTION, $fd))
  263. {
  264. StatisticClient::report(__CLASS__, 'CMD_ON_GATEWAY_CONNECTION', 0, 131, __CLASS__.'::accept()->sendToWorker() fail');
  265. }
  266. else
  267. {
  268. StatisticClient::report(__CLASS__, 'CMD_ON_GATEWAY_CONNECTION', 1, 0, '');
  269. }
  270. }
  271. return $new_connection;
  272. }
  273. /**
  274. * 存储全局的通信地址
  275. * @param string $address
  276. */
  277. protected function registerAddress($address)
  278. {
  279. // 统计打点
  280. StatisticClient::tick(__CLASS__, 'registerAddress');
  281. // 这里使用了信号量只能实现单机互斥,分布式互斥需要借助于memcached incr cas 或者其他分布式存储
  282. \Man\Core\Lib\Mutex::get();
  283. // key
  284. $key = 'GLOBAL_GATEWAY_ADDRESS';
  285. // 获取实例
  286. try
  287. {
  288. $store = Store::instance('gateway');
  289. }
  290. catch(\Exception $msg)
  291. {
  292. StatisticClient::report(__CLASS__, 'registerAddress', 0, 107, $msg);
  293. \Man\Core\Lib\Mutex::release();
  294. return false;
  295. }
  296. // 获取数据
  297. $addresses_list = $store->get($key);
  298. if(empty($addresses_list))
  299. {
  300. $addresses_list = array();
  301. }
  302. // 添加数据
  303. $addresses_list[$address] = $address;
  304. // 存储
  305. if(!$store->set($key, $addresses_list))
  306. {
  307. // 存储失败
  308. \Man\Core\Lib\Mutex::release();
  309. $msg = "注册gateway通信地址出错";
  310. if(get_class($store) == 'Memcached')
  311. {
  312. $msg .= " 原因:".$store->getResultMessage();
  313. }
  314. $this->notice($msg);
  315. StatisticClient::report(__CLASS__, 'registerAddress', 0, 107, new \Exception($msg));
  316. return false;
  317. }
  318. // 存储成功
  319. \Man\Core\Lib\Mutex::release();
  320. StatisticClient::report(__CLASS__, 'registerAddress', 1, 0, '');
  321. return true;
  322. }
  323. /**
  324. * 删除全局的通信地址
  325. * @param string $address
  326. */
  327. protected function unregisterAddress($address)
  328. {
  329. // 统计打点
  330. StatisticClient::tick(__CLASS__, 'unregisterAddress');
  331. // 这里使用了信号量只能实现单机互斥,分布式互斥需要借助于memcached incr cas或者其他分布式存储
  332. \Man\Core\Lib\Mutex::get();
  333. // key
  334. $key = 'GLOBAL_GATEWAY_ADDRESS';
  335. // 获取存储实例
  336. try
  337. {
  338. $store = Store::instance('gateway');
  339. }
  340. catch (\Exception $msg)
  341. {
  342. StatisticClient::report(__CLASS__, 'unregisterAddress', 0, 108, $msg);
  343. \Man\Core\Lib\Mutex::release();
  344. return false;
  345. }
  346. // 获取数据
  347. $addresses_list = $store->get($key);
  348. if(empty($addresses_list))
  349. {
  350. $addresses_list = array();
  351. }
  352. // 去掉要删除的数据
  353. unset($addresses_list[$address]);
  354. // 保存数据
  355. if(!$store->set($key, $addresses_list))
  356. {
  357. \Man\Core\Lib\Mutex::release();
  358. $msg = "删除gateway通信地址出错";
  359. if(get_class($store) == 'Memcached')
  360. {
  361. $msg .= " 原因:".$store->getResultMessage();
  362. }
  363. $this->notice($msg);
  364. StatisticClient::report(__CLASS__, 'unregisterAddress', 0, 108, new \Exception($msg));
  365. return;
  366. }
  367. // 存储成功
  368. \Man\Core\Lib\Mutex::release();
  369. StatisticClient::report(__CLASS__, 'unregisterAddress', 1, 0, '');
  370. }
  371. /**
  372. * 接收Udp数据
  373. * 如果数据超过一个udp包长,需要业务自己解析包体,判断数据是否全部到达
  374. * @param resource $socket
  375. * @param $null_one $flag
  376. * @param $null_two $base
  377. * @return void
  378. */
  379. public function recvInnerUdp($socket, $null_one = null, $null_two = null)
  380. {
  381. $data = stream_socket_recvfrom($socket , self::MAX_UDP_PACKEG_SIZE, 0, $address);
  382. // 惊群效应
  383. if(false === $data || empty($address))
  384. {
  385. return false;
  386. }
  387. $this->currentClientAddress = $address;
  388. $this->innerDealProcess($data);
  389. }
  390. /**
  391. * 内部通讯端口接受BusinessWorker连接请求,以便建立起长连接
  392. * @param resouce $socket
  393. * @param null $null_one
  394. * @param null $null_two
  395. */
  396. public function acceptInnerTcp($socket, $null_one = null, $null_two = null)
  397. {
  398. // 获得一个连接
  399. $new_connection = @stream_socket_accept($socket, 0);
  400. if(false === $new_connection)
  401. {
  402. return false;
  403. }
  404. // 连接的fd序号
  405. $fd = (int) $new_connection;
  406. $this->connections[$fd] = $new_connection;
  407. $this->recvBuffers[$fd] = array('buf'=>'', 'remain_len'=>GatewayProtocol::HEAD_LEN);
  408. // 非阻塞
  409. stream_set_blocking($this->connections[$fd], 0);
  410. $this->event->add($this->connections[$fd], \Man\Core\Events\BaseEvent::EV_READ , array($this, 'recvInnerTcp'), $fd);
  411. // 标记这个连接是内部通讯长连接,区别于客户端连接
  412. $this->workerConnections[$fd] = $fd;
  413. return $new_connection;
  414. }
  415. /**
  416. * 内部通讯判断数据是否全部到达
  417. * @param string $buffer
  418. */
  419. public function dealInnerInput($buffer)
  420. {
  421. return GatewayProtocol::input($buffer);
  422. }
  423. /**
  424. * 处理内部通讯收到的数据
  425. * @param event_buffer $event_buffer
  426. * @param int $fd
  427. * @return void
  428. */
  429. public function recvInnerTcp($connection, $flag, $fd = null)
  430. {
  431. $this->currentDealFd = $fd;
  432. $buffer = stream_socket_recvfrom($connection, $this->recvBuffers[$fd]['remain_len']);
  433. // 出错了
  434. if('' == $buffer && '' == ($buffer = fread($connection, $this->recvBuffers[$fd]['remain_len'])))
  435. {
  436. // 判断是否是链接断开
  437. if(!feof($connection))
  438. {
  439. return;
  440. }
  441. // 如果该链接对应的buffer有数据,说明发生错误
  442. if(!empty($this->recvBuffers[$fd]['buf']))
  443. {
  444. $this->statusInfo['send_fail']++;
  445. }
  446. // 关闭链接
  447. $this->closeInnerClient($fd);
  448. $this->notice("CLIENT:".$this->getRemoteIp()." CLOSE INNER_CONNECTION\n");
  449. if($this->workerStatus == self::STATUS_SHUTDOWN)
  450. {
  451. $this->stop();
  452. }
  453. return;
  454. }
  455. $this->recvBuffers[$fd]['buf'] .= $buffer;
  456. $remain_len = $this->dealInnerInput($this->recvBuffers[$fd]['buf']);
  457. // 包接收完毕
  458. if(0 === $remain_len)
  459. {
  460. // 内部通讯业务处理
  461. $this->innerDealProcess($this->recvBuffers[$fd]['buf']);
  462. $this->recvBuffers[$fd] = array('buf'=>'', 'remain_len'=>GatewayProtocol::HEAD_LEN);
  463. }
  464. // 出错
  465. else if(false === $remain_len)
  466. {
  467. // 出错
  468. $this->statusInfo['packet_err']++;
  469. $this->notice("INNER_PACKET_ERROR and CLOSE_INNER_CONNECTION\nCLIENT_IP:".$this->getRemoteIp()."\nBUFFER:[".bin2hex($this->recvBuffers[$fd]['buf'])."]\n");
  470. $this->closeInnerClient($fd);
  471. }
  472. else
  473. {
  474. $this->recvBuffers[$fd]['remain_len'] = $remain_len;
  475. }
  476. // 检查是否是关闭状态或者是否到达请求上限
  477. if($this->workerStatus == self::STATUS_SHUTDOWN )
  478. {
  479. // 停止服务
  480. $this->stop();
  481. // EXIT_WAIT_TIME秒后退出进程
  482. pcntl_alarm(self::EXIT_WAIT_TIME);
  483. }
  484. }
  485. /**
  486. * 内部通讯处理
  487. * @param string $recv_buffer
  488. */
  489. public function innerDealProcess($recv_buffer)
  490. {
  491. $pack = new GatewayProtocol($recv_buffer);
  492. $cmd = $pack->header['cmd'];
  493. $interface = isset(self::$interfaceMap[$cmd]) ? self::$interfaceMap[$cmd] : $cmd;
  494. StatisticClient::tick(__CLASS__, $interface);
  495. try
  496. {
  497. switch($cmd)
  498. {
  499. // 向某客户端发送数据
  500. case GatewayProtocol::CMD_SEND_TO_ONE:
  501. if(false === $this->sendToSocketId($pack->header['socket_id'], $pack->body))
  502. {
  503. throw new \Exception('发送数据到客户端失败,可能是该客户端的发送缓冲区已满,或者客户端已经下线', 121);
  504. }
  505. break;
  506. // 踢掉客户端
  507. case GatewayProtocol::CMD_KICK:
  508. $this->closeClient($pack->header['socket_id']);
  509. break;
  510. // 向所客户端发送数据
  511. case GatewayProtocol::CMD_SEND_TO_ALL:
  512. if($pack->ext_data)
  513. {
  514. $client_id_array = unpack('N*', $pack->ext_data);
  515. foreach($client_id_array as $client_id)
  516. {
  517. if(isset($this->clientConnMap[$client_id]))
  518. {
  519. $this->sendToSocketId($this->clientConnMap[$client_id], $pack->body);
  520. }
  521. }
  522. }
  523. else
  524. {
  525. $this->broadCast($pack->body);
  526. }
  527. break;
  528. // 更新某个客户端的session
  529. case GatewayProtocol::CMD_UPDATE_SESSION:
  530. if(isset($this->connSessionMap[$pack->header['socket_id']]))
  531. {
  532. $this->connSessionMap[$pack->header['socket_id']] = $pack->ext_data;
  533. }
  534. break;
  535. // 获得在线状态
  536. case GatewayProtocol::CMD_GET_ONLINE_STATUS:
  537. $online_status = json_encode(array_values($this->connClientMap));
  538. stream_socket_sendto($this->innerMainSocketUdp, $online_status, 0, $this->currentClientAddress);
  539. break;
  540. // 判断某个客户端id是否在线
  541. case GatewayProtocol::CMD_IS_ONLINE:
  542. stream_socket_sendto($this->innerMainSocketUdp, (int)isset($this->clientConnMap[$pack->header['client_id']]), 0, $this->currentClientAddress);
  543. break;
  544. // 未知的命令
  545. default :
  546. $err_msg = "gateway inner pack err cmd=$cmd";
  547. $this->notice($err_msg);
  548. throw new \Exception($err_msg, 110);
  549. }
  550. }
  551. catch(\Exception $msg)
  552. {
  553. StatisticClient::report(__CLASS__, $interface, 0, $msg->getCode() > 0 ? $msg->getCode() : 111, $msg);
  554. return;
  555. }
  556. StatisticClient::report(__CLASS__, $interface, 1, 0, '');
  557. }
  558. /**
  559. * 广播数据
  560. * @param string $bin_data
  561. */
  562. protected function broadCast($bin_data)
  563. {
  564. foreach($this->clientConnMap as $client_id=>$conn)
  565. {
  566. $this->sendToSocketId($conn, $bin_data);
  567. }
  568. }
  569. /**
  570. * 根据client_id获取client_id对应连接的socket_id
  571. * @param int $client_id
  572. */
  573. protected function getFdByClientId($client_id)
  574. {
  575. if(isset($this->clientConnMap[$client_id]))
  576. {
  577. return $this->clientConnMap[$client_id];
  578. }
  579. return 0;
  580. }
  581. /**
  582. * 根据连接socket_id获取client_id
  583. * @param int $fd
  584. */
  585. protected function getClientIdByFd($fd)
  586. {
  587. if(isset($this->connClientMap[$fd]))
  588. {
  589. return $this->connClientMap[$fd];
  590. }
  591. return 0;
  592. }
  593. /**
  594. * 向某个socket_id的连接发送消息
  595. * @param int $socket_id
  596. * @param string $bin_data
  597. */
  598. public function sendToSocketId($socket_id, $bin_data)
  599. {
  600. if(!isset($this->connections[$socket_id]))
  601. {
  602. return null;
  603. }
  604. $this->currentDealFd = $socket_id;
  605. return $this->sendToClient($bin_data);
  606. }
  607. /**
  608. * 用户客户端关闭连接时触发
  609. * @see Man\Core.SocketWorker::closeClient()
  610. */
  611. protected function closeClient($fd = null)
  612. {
  613. if($client_id = $this->getClientIdByFd($fd))
  614. {
  615. $this->sendToWorker(GatewayProtocol::CMD_ON_CLOSE, $fd);
  616. $this->deleteClientAddress($client_id);
  617. unset($this->clientConnMap[$client_id]);
  618. }
  619. unset($this->connClientMap[$fd], $this->connSessionMap[$fd], $this->connRemoteAddressMap[$fd]);
  620. parent::closeClient($fd);
  621. }
  622. /**
  623. * 内部通讯socket在BusinessWorker主动关闭连接时触发
  624. * @param int $fd
  625. */
  626. protected function closeInnerClient($fd)
  627. {
  628. unset($this->workerConnections[$fd]);
  629. parent::closeClient($fd);
  630. }
  631. /**
  632. * 随机抽取一个与BusinessWorker的长连接,将数据发给一个BusinessWorker
  633. * @param int $cmd
  634. * @param int $socket_id
  635. * @param string $body
  636. */
  637. protected function sendToWorker($cmd, $socket_id, $body = '')
  638. {
  639. $pack = new GatewayProtocol();
  640. $pack->header['cmd'] = $cmd;
  641. $pack->header['local_ip'] = $this->lanIp;
  642. $pack->header['local_port'] = $this->lanPort;
  643. $pack->header['socket_id'] = $socket_id;
  644. $pack->header['client_ip'] = $this->connRemoteAddressMap[$socket_id]['ip'];
  645. $pack->header['client_port'] = $this->connRemoteAddressMap[$socket_id]['port'];
  646. $pack->header['client_id'] = $this->getClientIdByFd($socket_id);
  647. $pack->body = $body;
  648. $pack->ext_data = $this->connSessionMap[$pack->header['socket_id']];
  649. return $this->sendBufferToWorker($pack->getBuffer());
  650. }
  651. /**
  652. * 随机抽取一个与BusinessWorker的长连接,将数据发给一个BusinessWorker
  653. * @param string $bin_data
  654. */
  655. protected function sendBufferToWorker($bin_data)
  656. {
  657. if($this->currentDealFd = array_rand($this->workerConnections))
  658. {
  659. if(false === $this->sendToClient($bin_data))
  660. {
  661. $msg = "sendBufferToWorker fail. May be the send buffer are overflow";
  662. $this->notice($msg);
  663. StatisticClient::report(__CLASS__, 'sendBufferToWorker', 0, 101, new \Exception($msg));
  664. return false;
  665. }
  666. }
  667. else
  668. {
  669. $msg = "sendBufferToWorker fail. the Connections between Gateway and BusinessWorker are not ready";
  670. $this->notice($msg);
  671. StatisticClient::report(__CLASS__, 'sendBufferToWorker', 0, 102, new \Exception($msg));
  672. return false;
  673. }
  674. StatisticClient::report(__CLASS__, 'sendBufferToWorker', 1, 0, '');
  675. }
  676. /**
  677. * 打印日志
  678. * @see Man\Core.AbstractWorker::notice()
  679. */
  680. protected function notice($str, $display=true)
  681. {
  682. $str = 'Worker['.get_class($this).']:'."$str";
  683. Man\Core\Lib\Log::add($str);
  684. if($display && Man\Core\Lib\Config::get('workerman.debug') == 1)
  685. {
  686. echo $str."\n";
  687. }
  688. }
  689. /**
  690. * 进程停止时,清除一些数据,忽略错误
  691. * @see Man\Core.SocketWorker::onStop()
  692. */
  693. public function onStop()
  694. {
  695. $this->unregisterAddress($this->lanIp.':'.$this->lanPort);
  696. foreach($this->connClientMap as $client_id)
  697. {
  698. Store::instance('gateway')->delete($client_id);
  699. }
  700. }
  701. /**
  702. * 创建全局唯一的id
  703. */
  704. protected function createGlobalClientId()
  705. {
  706. StatisticClient::tick(__CLASS__, 'createGlobalClientId');
  707. $global_socket_key = 'GLOBAL_SOCKET_ID_KEY';
  708. $store = Store::instance('gateway');
  709. $global_client_id = $store->increment($global_socket_key);
  710. if(!$global_client_id || $global_client_id > 2147483646)
  711. {
  712. $store->set($global_socket_key, 0);
  713. $global_client_id = $store->increment($global_socket_key);
  714. }
  715. if(!$global_client_id)
  716. {
  717. $msg = "生成全局client_id出错";
  718. if(get_class($store) == 'Memcached')
  719. {
  720. $msg .= " 原因:".$store->getResultMessage();
  721. }
  722. $this->notice($msg);
  723. StatisticClient::report(__CLASS__, 'createGlobalClientId', 0, 104, new \Exception($msg));
  724. return $global_client_id;
  725. }
  726. StatisticClient::report(__CLASS__, 'createGlobalClientId', 1, 0, '');
  727. return $global_client_id;
  728. }
  729. /**
  730. * 保存客户端内部通讯地址
  731. * @param int $global_client_id
  732. * @param string $address
  733. */
  734. protected function storeClientAddress($global_client_id, $address)
  735. {
  736. StatisticClient::tick(__CLASS__, 'storeClientAddress');
  737. if(!Store::instance('gateway')->set($global_client_id, $address))
  738. {
  739. $msg = "保存客户端通讯地址出错";
  740. if(get_class(Store::instance('gateway')) == 'Memcached')
  741. {
  742. $msg .= " 原因:".Store::instance('gateway')->getResultMessage();
  743. }
  744. $this->notice($msg);
  745. return StatisticClient::report(__CLASS__, 'storeClientAddress', 0, 105, new \Exception($msg));
  746. }
  747. StatisticClient::report(__CLASS__, 'storeClientAddress', 1, 0, '');
  748. }
  749. /**
  750. * 删除客户端内部通讯地址
  751. * @param int $global_client_id
  752. * @param string $address
  753. */
  754. protected function deleteClientAddress($global_client_id)
  755. {
  756. StatisticClient::tick(__CLASS__, 'deleteClientAddress');
  757. if(!Store::instance('gateway')->delete($global_client_id))
  758. {
  759. $msg = "删除客户端通讯地址出错";
  760. if(get_class(Store::instance('gateway')) == 'Memcached')
  761. {
  762. $msg .= " 原因:".Store::instance('gateway')->getResultMessage();
  763. }
  764. $this->notice($msg);
  765. return StatisticClient::report(__CLASS__, 'deleteClientAddress', 0, 106, new \Exception($msg));
  766. }
  767. StatisticClient::report(__CLASS__, 'deleteClientAddress', 1, 0, '');
  768. }
  769. /**
  770. * 向客户端发送心跳数据
  771. * 并把规定时间内没回应(客户端发来的任何数据都算是回应)的客户端踢掉
  772. */
  773. public function ping()
  774. {
  775. // 清理下线的链接
  776. foreach($this->pingInfo as $client_id=>$not_response_count)
  777. {
  778. // 已经下线的忽略
  779. if(!isset($this->clientConnMap[$client_id]))
  780. {
  781. unset($this->pingInfo[$client_id]);
  782. continue;
  783. }
  784. // 上次发送的心跳还没有回复次数大于限定值就断开
  785. if($this->pingNotResponseLimit > 0 && $not_response_count >= $this->pingNotResponseLimit)
  786. {
  787. $this->closeClient($this->clientConnMap[$client_id]);
  788. }
  789. }
  790. // 向所有链接发送心跳数据
  791. foreach($this->clientConnMap as $client_id=>$conn)
  792. {
  793. // 如果没设置ping的数据,则不发送心跳给客户端,但是要求客户端在规定的时间内(ping_interval*ping_not_response_limit)有请求发来,不然会断开
  794. if($this->pingData)
  795. {
  796. $this->sendToSocketId($conn, $this->pingData);
  797. }
  798. if(isset($this->pingInfo[$client_id]))
  799. {
  800. $this->pingInfo[$client_id]++;
  801. }
  802. else
  803. {
  804. $this->pingInfo[$client_id] = 1;
  805. }
  806. }
  807. }
  808. }