webserver.php 9.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232
  1. <?php
  2. require_once dirname(dirname(__FILE__)) . '/vendor/workerman/Autoloader.php'; //自动加载类
  3. require_once __DIR__ . '/config.php'; //配置文件
  4. use Workerman\Worker; //worker类
  5. use Workerman\Lib\Timer; //定时器函数
  6. //初始化 创建 websocket 服务器
  7. $workerWs = new Worker("websocket://127.0.0.1:12380");
  8. $workerWs->name = "WebSocket"; //命名为WebSocket
  9. $connection_count = 0;
  10. // 心跳检测的实现
  11. $workerWs->onWorkerStart = function ($workerWs) {
  12. global $connection_count;
  13. Timer::add(30, function () use ($workerWs) {
  14. $time_now = time();
  15. foreach ($workerWs->connections as $connection) {
  16. // 有可能该connection还没收到过消息,则lastMessageTime设置为当前时间
  17. if (empty($connection->lastMessageTime)) {
  18. $connection->lastMessageTime = $time_now;
  19. continue;
  20. }
  21. //上次通讯时间间隔大于心跳间隔,则认为客户端已经下线,关闭连接
  22. if ($time_now - $connection->lastMessageTime > HEARTBEAT_TIME) {
  23. $connection->close();
  24. }
  25. }
  26. });
  27. };
  28. // 服务器收到信息后执行
  29. /**
  30. * @param $connection
  31. * @param $data
  32. */
  33. $workerWs->onMessage = function ($connection, $data) {
  34. $connection->lastMessageTime = time(); //最后的信息时间
  35. global $connection_count;
  36. global $workerWs;
  37. $connection_count = count($workerWs->connections);
  38. $bufExplode = explode("\n", $data); //通过指定的分隔符,把字符串打散为数组
  39. foreach ($bufExplode as $key => $value) {
  40. $msg = json_decode($value, true); //解析数据
  41. if ($msg) {
  42. switch ($msg['type']) {
  43. case "heartbeat": //心跳检测(前端的心跳检测)
  44. $results['type'] = 'heartbeat';
  45. $results['data'] = 'I heard';
  46. // sendToWeb($msg);
  47. $results['connection_count'] = $connection_count;
  48. $connection->send(json_encode($results) . "\n");
  49. break;
  50. case "login": //异步客户端的登录(标识身份)
  51. $connection->uid = $msg['uid'];
  52. break;
  53. // 从前端传递过来的数据 开始位置======================================================================
  54. case "vue_get_order_record": //查询当前跟单订单
  55. sendToTcp([
  56. 'type' => 'get_order_record',
  57. 'order' => $msg['orderid']
  58. ]);
  59. break;
  60. case "vue_set_current_order": //设置当前跟单订单信息
  61. sendToTcp([
  62. 'type' => 'set_current_order',
  63. 'order' => (int)$msg['orderid']
  64. ]);
  65. break;
  66. case "vue_insert": //插入发送内容
  67. if(isset($msg['message'])){
  68. sendToTcp([
  69. 'type' => 'insert',
  70. 'message' => 'ErrorInsert',
  71. 'orderid' => $msg['orderid'],
  72. 'data'=> $msg['data'],
  73. ]);
  74. }else{
  75. sendToTcp([
  76. 'type' => 'insert',
  77. 'orderid' => $msg['orderid'],
  78. 'data'=> $msg['data'],
  79. ]);
  80. }
  81. break;
  82. case "vue_rollback": //Rollback请求
  83. sendToTcp([
  84. 'type' => 'rollback',
  85. 'data'=> $msg['data'],
  86. ]);
  87. break;
  88. case "vue_except_rollback": //Rollback请求
  89. sendToTcp([
  90. 'type' => 'except_rollback',
  91. 'data'=> $msg['data'],
  92. ]);
  93. break;
  94. case "vue_retry": //Retry请求
  95. sendToTcp([
  96. 'type' => 'retry',
  97. 'data'=> $msg['data'],
  98. ]);
  99. break;
  100. //从前端传递过来的数据 结束位置==================================================================
  101. //从接口过来的数据 开始位置================(这里的数据是一条一条返回的)==============================================
  102. case "get_order_record": //查询当前跟单订单
  103. sendToWeb([
  104. 'type' => $msg['type'],
  105. 'err_code' => $msg['err_code'],
  106. 'login' => $msg['login'],
  107. 'order' => $msg['order'],
  108. 'symbol' => $msg['symbol'],
  109. 'digit' => $msg['digit'],
  110. 'volume' => $msg['volume'],
  111. 'open_time' => $msg['open_time'],
  112. 'open_price' => $msg['open_price'],
  113. 'close_time' => $msg['close_time'],
  114. 'close_price'=> $msg['close_price'],
  115. 'sl' => $msg['sl'],
  116. 'tp' => $msg['tp'],
  117. 'profit' => $msg['profit'],
  118. 'commission' => $msg['commission'],
  119. 'swap' => $msg['swap'],
  120. 'comment' => $msg['comment'],
  121. ]);
  122. break;
  123. case "set_current_order": //设置当前跟单订单信息
  124. sendToWeb([
  125. 'type' => $msg['type'],
  126. 'error_code' => $msg['error_code']
  127. ]);
  128. break;
  129. case "insert": //插入发送内容
  130. sendToWeb([
  131. 'type' => $msg['type'],
  132. 'orig_order' => $msg['orig_order'],
  133. 'orig_login' => $msg['orig_login'],
  134. 'dest' => $msg['dest']
  135. ]);
  136. break;
  137. case "rollback": //Rollback请求
  138. sendToWeb([
  139. 'type' => $msg['type'],
  140. 'orig_order' => $msg['orig_order'],
  141. 'orig_login' => $msg['orig_login'],
  142. 'desc' => $msg['desc']
  143. ]);
  144. break;
  145. case "except_rollback": //retry表中 Rollback请求
  146. sendToWeb([
  147. 'type' => $msg['type'],
  148. 'orig_order' => $msg['orig_order'],
  149. 'orig_login' => $msg['orig_login'],
  150. 'desc' => $msg['desc']
  151. ]);
  152. break;
  153. case "retry": //Retry请求
  154. sendToWeb([
  155. 'type' => $msg['type'],
  156. 'orig_order' => $msg['orig_order'], //最初的订单
  157. 'orig_login' => $msg['orig_login'], //最初的login
  158. 'desc' => $msg['desc']
  159. ]);
  160. break;
  161. //从接口过来的数据 结束位置========================(这里的数据是一条一条返回的)==============================================
  162. default:
  163. $results['type'] = 'info';
  164. $results['data'] = '未知类型';
  165. $connection->send(json_encode($results) . "\n");
  166. break;
  167. }
  168. } /*else {
  169. // 没有信息的时候
  170. $results['type'] = 'info';
  171. $results['data'] = 'json格式错误';
  172. $connection->send(json_encode($results) . "\n");
  173. }*/
  174. }
  175. };
  176. // 向所有的web用户发送 ( vue前端的 客户端 这里的客户端是没有uid的)
  177. function sendToWeb($msg)
  178. {
  179. global $workerWs;
  180. foreach ($workerWs->connections as $connection) {
  181. if (!isset($connection->uid)) {
  182. $connection->send(json_encode($msg) . "\n");
  183. }
  184. }
  185. }
  186. // 向本机的tcp服务器发送请求(tcp上有一个异步请求的客户端 他登陆是有uid字段的赋值的 )
  187. function sendToTcp($msg)
  188. {
  189. global $workerWs;
  190. foreach ($workerWs->connections as $connection) {
  191. if (isset($connection->uid) && $connection->uid == "tcp") { //只针对特定的用户发起请求
  192. $connection->send(json_encode($msg) . "\n");
  193. }
  194. }
  195. }
  196. Worker::runAll(); //执行函数