微信搜索superit|邀请体验:大数据, 数据管理、OLAP分析与可视化平台 | 赞助作者:赞助作者

【workerman源码分析】异步TCP连接AsyncTcpConnection.php

php aide_941 24℃

【workerman源码分析】异步TCP连接AsyncTcpConnection.php

版权声明:本文为博主原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。

本文链接:https://blog.csdn.net/qq2942713658/article/details/81945323
  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\Connection;
  15. use Workerman\Events\EventInterface;
  16. use Workerman\Lib\Timer;
  17. use Workerman\Worker;
  18. use Exception;
  19. /**
  20. * 异步TCP连接
  21. */
  22. class AsyncTcpConnection extends TcpConnection
  23. {
  24. /**
  25. * 成功建立socket连接时发出。
  26. *
  27. * @var callback
  28. */
  29. public $onConnect = null;
  30. /**
  31. * 传输层协议
  32. *
  33. * @var string
  34. */
  35. public $transport = ‘tcp’;
  36. /**
  37. * 状态
  38. *
  39. * @var int
  40. */
  41. protected $_status = self::STATUS_INITIAL;
  42. /**
  43. * 远程主机
  44. *
  45. * @var string
  46. */
  47. protected $_remoteHost = ;
  48. /**
  49. * 远程端口
  50. *
  51. * @var int
  52. */
  53. protected $_remotePort = 80;
  54. /**
  55. * 连接开始时间
  56. *
  57. * @var string
  58. */
  59. protected $_connectStartTime = 0;
  60. /**
  61. * 远程URI
  62. *
  63. * @var string
  64. */
  65. protected $_remoteURI = ;
  66. /**
  67. * 文本选项
  68. *
  69. * @var array
  70. */
  71. protected $_contextOption = null;
  72. /**
  73. * 重新连接定时器
  74. *
  75. * @var int
  76. */
  77. protected $_reconnectTimer = null;
  78. /**
  79. * PHP内置协议
  80. *
  81. * @var array
  82. */
  83. protected static $_builtinTransports = array(
  84. ‘tcp’ => ‘tcp’,
  85. ‘udp’ => ‘udp’,
  86. ‘unix’ => ‘unix’,
  87. ‘ssl’ => ‘ssl’,
  88. ‘sslv2’ => ‘sslv2’,
  89. ‘sslv3’ => ‘sslv3’,
  90. ‘tls’ => ‘tls’
  91. );
  92. /**
  93. * 构造方法
  94. *
  95. * @param string $remote_address
  96. * @param array $context_option
  97. * @throws Exception
  98. */
  99. public function __construct($remote_address, $context_option = null)
  100. {
  101. $address_info = parse_url($remote_address);
  102. if (!$address_info) {
  103. list($scheme, $this->_remoteAddress) = explode(‘:’, $remote_address, 2);
  104. if (!$this->_remoteAddress) {
  105. Worker::safeEcho(new \Exception(‘bad remote_address’));
  106. }
  107. } else {
  108. if (!isset($address_info[‘port’])) {
  109. $address_info[‘port’] = 80;
  110. }
  111. if (!isset($address_info[‘path’])) {
  112. $address_info[‘path’] = ‘/’;
  113. }
  114. if (!isset($address_info[‘query’])) {
  115. $address_info[‘query’] = ;
  116. } else {
  117. $address_info[‘query’] = ‘?’ . $address_info[‘query’];
  118. }
  119. $this->_remoteAddress = “{$address_info[‘host’]}:{$address_info[‘port’]}”;
  120. $this->_remoteHost = $address_info[‘host’];
  121. $this->_remotePort = $address_info[‘port’];
  122. $this->_remoteURI = “{$address_info[‘path’]}{$address_info[‘query’]}”;
  123. $scheme = isset($address_info[‘scheme’]) ? $address_info[‘scheme’] : ‘tcp’;
  124. }
  125. $this->id = $this->_id = self::$_idRecorder++;
  126. if(PHP_INT_MAX === self::$_idRecorder){
  127. self::$_idRecorder = 0;
  128. }
  129. // 检查应用层协议类。
  130. if (!isset(self::$_builtinTransports[$scheme])) {
  131. $scheme = ucfirst($scheme);
  132. $this->protocol = ‘\\Protocols\\’ . $scheme;
  133. if (!class_exists($this->protocol)) {
  134. $this->protocol = “\\Workerman\\Protocols\\$scheme”;
  135. if (!class_exists($this->protocol)) {
  136. throw new Exception(“class \\Protocols\\$scheme not exist”);
  137. }
  138. }
  139. } else {
  140. $this->transport = self::$_builtinTransports[$scheme];
  141. }
  142. // 统计
  143. self::$statistics[‘connection_count’]++;
  144. $this->maxSendBufferSize = self::$defaultMaxSendBufferSize;
  145. $this->_contextOption = $context_option;
  146. static::$connections[$this->_id] = $this;
  147. }
  148. /**
  149. * 连接
  150. *
  151. * @return void
  152. */
  153. public function connect()
  154. {
  155. if ($this->_status !== self::STATUS_INITIAL && $this->_status !== self::STATUS_CLOSING &&
  156. $this->_status !== self::STATUS_CLOSED) {
  157. return;
  158. }
  159. $this->_status = self::STATUS_CONNECTING;
  160. $this->_connectStartTime = microtime(true);
  161. if ($this->transport !== ‘unix’) {
  162. // 异步打开socket连接
  163. if ($this->_contextOption) {
  164. $context = stream_context_create($this->_contextOption);
  165. $this->_socket = stream_socket_client(“tcp://{$this->_remoteHost}:{$this->_remotePort}”,
  166. $errno, $errstr, 0, STREAM_CLIENT_ASYNC_CONNECT, $context);
  167. } else {
  168. $this->_socket = stream_socket_client(“tcp://{$this->_remoteHost}:{$this->_remotePort}”,
  169. $errno, $errstr, 0, STREAM_CLIENT_ASYNC_CONNECT);
  170. }
  171. } else {
  172. $this->_socket = stream_socket_client(“{$this->transport}://{$this->_remoteAddress}”, $errno, $errstr, 0,
  173. STREAM_CLIENT_ASYNC_CONNECT);
  174. }
  175. // 如果失败尝试发出onError回调。
  176. if (!$this->_socket) {
  177. $this->emitError(WORKERMAN_CONNECT_FAIL, $errstr);
  178. if ($this->_status === self::STATUS_CLOSING) {
  179. $this->destroy();
  180. }
  181. if ($this->_status === self::STATUS_CLOSED) {
  182. $this->onConnect = null;
  183. }
  184. return;
  185. }
  186. // 将socket添加到全局事件循环等待连接已成功建立或失败。
  187. Worker::$globalEvent->add($this->_socket, EventInterface::EV_WRITE, array($this, ‘checkConnection’));
  188. // For windows.
  189. if(DIRECTORY_SEPARATOR === ‘\\’) {
  190. Worker::$globalEvent->add($this->_socket, EventInterface::EV_EXCEPT, array($this, ‘checkConnection’));
  191. }
  192. }
  193. /**
  194. * 重新连接
  195. *
  196. * @param int $after
  197. * @return void
  198. */
  199. public function reconnect($after = 0)
  200. {
  201. $this->_status = self::STATUS_INITIAL;
  202. static::$connections[$this->_id] = $this;
  203. if ($this->_reconnectTimer) {
  204. Timer::del($this->_reconnectTimer);
  205. }
  206. if ($after > 0) {
  207. $this->_reconnectTimer = Timer::add($after, array($this, ‘connect’), null, false);
  208. return;
  209. }
  210. $this->connect();
  211. }
  212. /**
  213. * 取消重新连接
  214. */
  215. public function cancelReconnect()
  216. {
  217. if ($this->_reconnectTimer) {
  218. Timer::del($this->_reconnectTimer);
  219. }
  220. }
  221. /**
  222. * 获取远程地址
  223. *
  224. * @return string
  225. */
  226. public function getRemoteHost()
  227. {
  228. return $this->_remoteHost;
  229. }
  230. /**
  231. * 获取远程URI.
  232. *
  233. * @return string
  234. */
  235. public function getRemoteURI()
  236. {
  237. return $this->_remoteURI;
  238. }
  239. /**
  240. * 尝试发出onError回调
  241. *
  242. * @param int $code
  243. * @param string $msg
  244. * @return void
  245. */
  246. protected function emitError($code, $msg)
  247. {
  248. $this->_status = self::STATUS_CLOSING;
  249. if ($this->onError) {
  250. try {
  251. call_user_func($this->onError, $this, $code, $msg);
  252. } catch (\Exception $e) {
  253. Worker::log($e);
  254. exit(250);
  255. } catch (\Error $e) {
  256. Worker::log($e);
  257. exit(250);
  258. }
  259. }
  260. }
  261. /**
  262. * 检查连接是否成功建立
  263. *
  264. * @param resource $socket
  265. * @return void
  266. */
  267. public function checkConnection()
  268. {
  269. if ($this->_status != self::STATUS_CONNECTING) {
  270. return;
  271. }
  272. // 删除Windows的EV_EXPECT
  273. if(DIRECTORY_SEPARATOR === ‘\\’) {
  274. Worker::$globalEvent->del($this->_socket, EventInterface::EV_EXCEPT);
  275. }
  276. // 检查socket状态
  277. if ($address = stream_socket_get_name($this->_socket, true)) {
  278. // 非阻塞
  279. stream_set_blocking($this->_socket, 0);
  280. // 适配 hhvm
  281. if (function_exists(‘stream_set_read_buffer’)) {
  282. stream_set_read_buffer($this->_socket, 0);
  283. }
  284. // 尝试打开tcp的keepalive并禁用Nagle算法
  285. if (function_exists(‘socket_import_stream’) && $this->transport === ‘tcp’) {
  286. $raw_socket = socket_import_stream($this->_socket);
  287. socket_set_option($raw_socket, SOL_SOCKET, SO_KEEPALIVE, 1);
  288. socket_set_option($raw_socket, SOL_TCP, TCP_NODELAY, 1);
  289. }
  290. // 删除写侦听器
  291. Worker::$globalEvent->del($this->_socket, EventInterface::EV_WRITE);
  292. // SSL握手
  293. if ($this->transport === ‘ssl’) {
  294. $this->_sslHandshakeCompleted = $this->doSslHandshake($this->_socket);
  295. } else {
  296. // 等待发送的数据
  297. if ($this->_sendBuffer) {
  298. Worker::$globalEvent->add($this->_socket, EventInterface::EV_WRITE, array($this, ‘baseWrite’));
  299. }
  300. }
  301. // 注册侦听器等待读取事件。
  302. Worker::$globalEvent->add($this->_socket, EventInterface::EV_READ, array($this, ‘baseRead’));
  303. $this->_status = self::STATUS_ESTABLISHED;
  304. $this->_remoteAddress = $address;
  305. // 调用onConnect回调
  306. if ($this->onConnect) {
  307. try {
  308. call_user_func($this->onConnect, $this);
  309. } catch (\Exception $e) {
  310. Worker::log($e);
  311. exit(250);
  312. } catch (\Error $e) {
  313. Worker::log($e);
  314. exit(250);
  315. }
  316. }
  317. // 调用protocol::onConnect
  318. if (method_exists($this->protocol, ‘onConnect’)) {
  319. try {
  320. call_user_func(array($this->protocol, ‘onConnect’), $this);
  321. } catch (\Exception $e) {
  322. Worker::log($e);
  323. exit(250);
  324. } catch (\Error $e) {
  325. Worker::log($e);
  326. exit(250);
  327. }
  328. }
  329. } else {
  330. // 连接失败
  331. $this->emitError(WORKERMAN_CONNECT_FAIL, ‘connect ‘ . $this->_remoteAddress . ‘ fail after ‘ . round(microtime(true) – $this->_connectStartTime, 4) . ‘ seconds’);
  332. if ($this->_status === self::STATUS_CLOSING) {
  333. $this->destroy();
  334. }
  335. if ($this->_status === self::STATUS_CLOSED) {
  336. $this->onConnect = null;
  337. }
  338. }
  339. }
  340. }

 

转载请注明:SuperIT » 【workerman源码分析】异步TCP连接AsyncTcpConnection.php

喜欢 (0)or分享 (0)