tcpServer.php 31 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899
  1. <?php
  2. require dirname(dirname(__FILE__)).'/vendor/autoload.php';
  3. require_once dirname(dirname(__FILE__)) . '/vendor/workerman/Autoloader.php';
  4. require_once dirname(dirname(__FILE__)) . '/vendor/mysql/src/Connection.php';
  5. require_once __DIR__ . '/config.php';
  6. use Workerman\Worker; //worker容器类
  7. use Workerman\Connection\AsyncTcpConnection; //连接异步客户端
  8. use Workerman\MySQL\Connection; //连接数据库的类
  9. use Workerman\Lib\Timer; //定时器函数
  10. use Monolog\Logger;
  11. use Monolog\Handler\StreamHandler;
  12. use Monolog\Handler\ErrorLogHandler;
  13. date_default_timezone_set("PRC");
  14. //初始化 (创建了一个tcp服务)
  15. $workerTcp = new Worker("tcp://127.0.0.1:12345");
  16. $workerTcp->name = "TCP";
  17. $workerTcp::$logFile = __DIR__.'/workerman.log';
  18. $num = 1;
  19. $workerTcp->onWorkerStart = function ($workerTcp) {
  20. global $db; //数据库
  21. global $con; //访问本地创建的webserver服务器
  22. global $connection_to_tcp; //访问接口服务器(远程的)
  23. global $connection_count;
  24. global $send_data; //存储insert数据
  25. global $index; //insert管理数据序号
  26. global $send_order; //insert全局的订单号
  27. global $seqnum;
  28. global $message;
  29. global $rollback_data; //rollback数据
  30. global $rollback_index; //rollback 管理数据序号
  31. global $retry_data; //retry数据
  32. global $retry_index; //retry 管理数据的序号
  33. global $retry_rollback_data; //retry表中rollback数据
  34. global $retry_rollback_index; //retry表rollback管理数据的序号
  35. global $token; //储存token
  36. $message = 0;
  37. $seqnum = 1;
  38. getToken();
  39. $db = new Connection(DB_HOST, DB_PORT, DB_USERNAME, DB_PASSWORD, DB_NAME); //本地数据库的配置
  40. //访问本地的websocket服务器=============================================
  41. $con = new AsyncTcpConnection("ws://127.0.0.1:12380");
  42. $con->onConnect = function ($on) {
  43. global $index;
  44. global $rollback_index;
  45. global $retry_index;
  46. global $retry_rollback_index;
  47. $index = 0;
  48. $rollback_index = 0;
  49. $retry_index = 0;
  50. $retry_rollback_index = 0;
  51. sendToWebServer([
  52. 'type' => 'login',
  53. 'uid' => 'tcp'
  54. ]);
  55. };
  56. // websocke服务器发送信息过来触发的函数
  57. $con->onMessage = function ($con, $data) {
  58. global $send_data;
  59. global $connection_count;
  60. $msg = json_decode($data, true);
  61. if ($msg['type'] == "heartbeat") {
  62. $connection_count = $msg["connection_count"];
  63. }
  64. if ($msg['type'] == "get_order_record") { //查询订单状态
  65. sendTo_tcp_Server($msg);
  66. }
  67. if ($msg['type'] == "set_current_order") { //设置当前跟单订单
  68. sendTo_tcp_Server($msg);
  69. }
  70. if($msg['type'] == 'set_current_position'){
  71. sendTo_tcp_Server($msg);
  72. }
  73. if ($msg['type'] == "insert") { //Insert请求
  74. global $index; //传递的index
  75. global $send_data; //全局的数据
  76. global $send_order; //全局的订单号码
  77. global $message;
  78. if(isset($msg['message'])){
  79. $message = 1;
  80. unset($msg['message']);
  81. }else{
  82. $message = 0;
  83. }
  84. $insert_data = $msg['data'];
  85. $insert_data = array_chunk($insert_data, 12); //每十个是一个数组
  86. $send_data = $insert_data; //传递的数据
  87. $order = trim($msg['orderid']);
  88. $send_order = (int)$order;
  89. send_insert($send_data, $index, $send_order);
  90. }
  91. if ($msg['type'] == "rollback") { //Rollback请求
  92. global $rollback_index; //传递index
  93. global $rollback_data; //全局的数据
  94. $rollback_data = array_chunk($msg['data'], 5); //每几个是一个数组
  95. send_rollback($rollback_data, $rollback_index);
  96. }
  97. if ($msg['type'] == "except_rollback") { //retry表中的 Rollback请求
  98. global $retry_rollback_data; //retry表中rollback数据
  99. global $retry_rollback_index; //retry表rollback管理数据的序号
  100. $retry_rollback_data = array_chunk($msg['data'], 5);
  101. send_retry_rollback($retry_rollback_data, $retry_rollback_index);
  102. }
  103. if ($msg['type'] == "retry") { //Retry请求
  104. global $retry_data; //retry数据
  105. global $retry_index; //retry 管理数据的序号
  106. $retry_data = array_chunk($msg['data'], 5);
  107. send_retry($retry_data, $retry_index);
  108. }
  109. };
  110. $con->onClose = function ($con) {
  111. echo "connection closed\n";
  112. $con->reConnect(1);
  113. };
  114. $con->onError = function ($con, $code, $msg) {
  115. echo "Error code:$code msg:$msg\n";
  116. };
  117. $con->connect();
  118. // 访问本地的websocket服务器=======================================================
  119. // 向远处的服务器发送信息===========================================================
  120. //$connection_to_tcp = new AsyncTcpConnection('tcp://119.23.51.113:10008');
  121. //$connection_to_tcp = new AsyncTcpConnection('tcp://127.0.0.1:1235');
  122. //$connection_to_tcp = new AsyncTcpConnection('tcp://192.168.5.111:10009');
  123. $connection_to_tcp = new AsyncTcpConnection('tcp://103.230.218.164:10008');
  124. $connection_to_tcp->onConnect = function ($connection_to_tcp) {
  125. // 发送心跳信息
  126. $connection_to_tcp->send('{"type":"ping"}' . "\n");
  127. };
  128. // 接口服务器向我发送的信息
  129. $connection_to_tcp->onMessage = function ($connection_to_tcp, $data) {
  130. tcpMessHandle($data);
  131. };
  132. $connection_to_tcp->onClose = function($connection_to_tcp)
  133. {
  134. echo "connection closed\n";
  135. log_file("to_tcp connection closed\n");
  136. // $connection_to_tcp->reConnect(); //执行连接
  137. };
  138. $connection_to_tcp->onError = function($connection_to_tcp, $code, $msg)
  139. {
  140. echo "Error code:$code msg:$msg\n";
  141. };
  142. // 向远处的服务器发送信息============================================================
  143. $connection_to_tcp->connect(); //执行连接
  144. //发送给 接口tcp服务器 心跳========================================================
  145. Timer::add(30, function () use ($connection_to_tcp) {
  146. //定时发送心跳信息
  147. $connection_to_tcp->send('{"type":"ping"}' . "\n");
  148. });
  149. // 发送给 webserver 服务器 心跳=====================================================
  150. Timer::add(1, function () use ($con,$connection_to_tcp) {
  151. // 定时发送心跳信息
  152. global $connection_count;
  153. if($connection_count){
  154. if($connection_count>1){
  155. $connection_to_tcp->connect(); //客户端上线重新连接接口
  156. $connection_to_tcp->onConnect = function ($connection_to_tcp) {
  157. // 发送心跳信息
  158. $connection_to_tcp->send('{"type":"ping"}' . "\n");
  159. };
  160. // 接口服务器向我发送的信息
  161. $connection_to_tcp->onMessage = function ($connection_to_tcp, $data) {
  162. tcpMessHandle($data);
  163. };
  164. $connection_to_tcp->onClose = function($connection_to_tcp)
  165. {
  166. echo "connection closed\n";
  167. log_file("to_tcp connection closed\n");
  168. // $connection_to_tcp->reConnect(); //执行连接
  169. };
  170. $connection_to_tcp->onError = function($connection_to_tcp, $code, $msg)
  171. {
  172. echo "Error code:$code msg:$msg\n";
  173. };
  174. }else{
  175. //客户端不在线
  176. $connection_to_tcp->close();
  177. }
  178. }
  179. $con->send(json_encode(['type' => 'heartbeat']). "\n");
  180. });
  181. Timer::add(300,function(){
  182. refreshToken();
  183. global $token;
  184. var_dump($token);
  185. });
  186. // 心跳检测(TCP 所有客户端) 心跳===================================================
  187. Timer::add(1, function () use ($workerTcp) {
  188. $time_now = time(); //当前时间
  189. foreach ($workerTcp->connections as $connection) {
  190. // 有可能该connection还没收到过消息,则lastMessageTime设置为当前时间
  191. if (empty($connection->lastMessageTime)) {
  192. $connection->lastMessageTime = $time_now;
  193. continue;
  194. }
  195. // 上次通讯时间间隔大于心跳间隔,则认为客户端已经下线,关闭连接
  196. if ($time_now - $connection->lastMessageTime > HEARTBEAT_TIME) {
  197. $connection->close();
  198. }
  199. }
  200. });
  201. };
  202. $workerTcp->onConnect = function ($connection) {
  203. echo "connection onConnect\n";
  204. };
  205. $workerTcp->onError = function ($connection, $code, $msg) {
  206. echo "Error code:$code msg:$msg\n";
  207. };
  208. $workerTcp->onClose = function ($connection) {
  209. echo "connection closed\n";
  210. };
  211. $workerTcp->onMessage = function ($connection, $data) {
  212. echo "connection onMessage\n";
  213. };
  214. //远程服务器返回消息处理
  215. function tcpMessHandle($data)
  216. {
  217. global $num;
  218. //这里显示的是从接口服务器返回的数据(需要根据数据的类似判断)
  219. global $db; //操作数据的全局变量
  220. global $recv_buffer; //存储不完整的数据
  221. global $index; //全局的变量
  222. global $send_data; //全局的数据
  223. global $send_order; //全局的订单号码
  224. global $rollback_data; //rollback数据
  225. global $rollback_index; //rollback 管理数据序号
  226. global $retry_data; //retry数据
  227. global $retry_index; //retry 管理数据的序号
  228. global $retry_rollback_data; //retry表中rollback数据
  229. global $retry_rollback_index; //retry表rollback管理数据的序号
  230. global $seqnum;
  231. global $message;
  232. $bufExplode = explode("\n", $data); //通过指定的分隔符,把字符串打散为数组
  233. foreach ($bufExplode as $key => $value) {
  234. if ($value) {
  235. $msg = json_decode($value, true); //对信息进行处理
  236. if (!$msg) {
  237. $recv_buffer .= $value; //把数据组装起来
  238. $msg1 = json_decode($recv_buffer, true);
  239. if ($msg1) {
  240. $msg = $msg1;
  241. } else {
  242. continue; //解析的不对就跳出循环
  243. }
  244. }
  245. } else {
  246. continue; //信息不存在就跳出循环
  247. }
  248. // 接口服务器返回的信息
  249. echo "接口返回数据";
  250. var_dump($msg);
  251. if ($msg['type'] == "get_order_record") { //查询订单状态
  252. $recv_buffer = "";
  253. /* if (!isset($msg['seqnum'])) {
  254. return false;
  255. }
  256. ack($msg['seqnum']);
  257. $seqnum++;*/
  258. }
  259. if ($msg['type'] == "set_current_order") { //设置当前跟单订单
  260. $recv_buffer = "";
  261. /*if (!isset($msg['seqnum'])) {
  262. return false;
  263. }*/
  264. sendToWebServer($msg);
  265. /* ack($msg['seqnum']);
  266. $seqnum++;*/
  267. }
  268. if ($msg['type'] == "insert") { //Insert请求 (有数据库操作,插入)
  269. $recv_buffer = "";
  270. /* if (!isset($msg['seqnum'])) {
  271. return false;
  272. }
  273. if ($seqnum != $msg["seqnum"]) {
  274. return false;
  275. }*/
  276. global $message;
  277. global $token;
  278. $index++; //将数据加一
  279. $total = count($send_data); //总数据的长度
  280. $insert_data = $msg['dest'];
  281. /* ack($msg['seqnum']);
  282. $seqnum++;*/
  283. if ($index < $total) {
  284. send_insert($send_data, $index, $send_order);
  285. } else {
  286. $index = 0;
  287. }
  288. $login = '';
  289. $arr = "";
  290. $time = time();
  291. $delete_sql = 'delete from order_progress where ';
  292. foreach ($insert_data as $key => $value) {
  293. if ($value['error_code'] == -2) {
  294. return false;
  295. } else {
  296. if($message == 1){
  297. if($value['error_code'] == 0){
  298. $arr .= "(" . "'{$msg['type']}'" . "," . $msg['orig_order'] . "," . $msg['orig_login'] . "," . $value['dest_login'] . "," . $value['dest_order'] . "," . $value['percentage'] . "," . $value['profit'] . "," . $value['error_code'] . "," . $time . ")" . ",";
  299. $delete_sql .= "(dest_login = $value[dest_login] and orig_order = $msg[orig_order] and error_code != 0) or ";
  300. $login .= $value['dest_login'].",";
  301. }
  302. }else{
  303. $arr .= "(" . "'{$msg['type']}'" . "," . $msg['orig_order'] . "," . $msg['orig_login'] . "," . $value['dest_login'] . "," . $value['dest_order'] . "," . $value['percentage'] . "," . $value['profit'] . "," . $value['error_code'] . "," . $time . ")" . ",";
  304. if($value['error_code'] == 0){
  305. $login .= $value['dest_login'].",";
  306. }
  307. }
  308. }
  309. }
  310. $filed = "type,orig_order,orig_login,dest_login,dest_order,percentage,profit,error_code,addtime";
  311. $arr = substr($arr, 0, -1);
  312. $sql = sprintf("INSERT INTO %s(%s) VALUES %s ", "order_progress", $filed, $arr);
  313. $sql2 = sprintf("INSERT INTO %s(%s) VALUES %s", "order_save", $filed, $arr);
  314. if($message == 1){
  315. if ($delete_sql != 'delete from order_progress where ') {
  316. $delete_sql = trim($delete_sql, 'or ');
  317. $db->query($delete_sql);
  318. $db->query($sql);
  319. $db->query($sql2);
  320. }
  321. }else{
  322. $db->query($sql);
  323. $db->query($sql2);
  324. }
  325. if($login){
  326. $logins = substr($login,0,-1);
  327. balancefix($logins);
  328. }
  329. /* if($message == 1){
  330. $msg['type'] = "ErrorInsert";
  331. }else{
  332. $msg['type'] = "insert";
  333. }*/
  334. sendToWebServer($msg); //发送给前端
  335. }
  336. if ($msg['type'] == "rollback") { //Rollback请求 (有数据库操作,插入并看看是否需要直接返回给前端)
  337. $recv_buffer = "";
  338. /* if (!isset($msg['seqnum'])) {
  339. return false;
  340. }
  341. if ($seqnum != $msg['seqnum']) {
  342. return false;
  343. }*/
  344. $rollback_index++; //将数据加一
  345. $total = count($rollback_data); //总数据的长度
  346. $rollbackdata = $msg['desc'];
  347. /* ack($msg["seqnum"]);
  348. $seqnum++;*/
  349. if ($rollback_index < $total) {
  350. send_rollback($rollback_data, $rollback_index);
  351. }else{
  352. $rollback_index = 0;
  353. }
  354. sendToWebServer($msg); //发送给前端
  355. $delete_sql = 'delete from order_progress where ';
  356. $insert_sql = "insert into order_save(type,orig_order,orig_login,dest_login,dest_orders,dest_deals,percentage,error_code,addtime) values";
  357. $update = [];
  358. $login = '';
  359. foreach ($rollbackdata as $key => $value) {
  360. if ($value['error_code'] == -2) {
  361. return false;
  362. } else {
  363. $time = time();
  364. if ($value['error_code'] == 0) {
  365. $delete_sql .= "(dest_login = $value[dest_login] and orig_order = $msg[orig_order]) or ";
  366. $value['dest_orders'] = json_encode($value['dest_orders']);
  367. $value['dest_deals'] = json_encode($value['dest_deals']);
  368. $insert_sql .= "('$msg[type]',$msg[orig_order],$msg[orig_login],$value[dest_login], $value[dest_orders],$value[dest_deals],$value[percentage],$value[error_code],$time),";
  369. $login .= $value['dest_login'].",";
  370. } else {
  371. //不成功把参数组装数组以便批量修改
  372. $rollbackdata[$key]['type'] = $msg['type'];
  373. $rollbackdata[$key]['orig_order'] = $msg['orig_order'];
  374. $rollbackdata[$key]['orig_login'] = $msg['orig_login'];
  375. $rollbackdata[$key]['addtime'] = time();
  376. array_push($update, $value);
  377. }
  378. }
  379. }
  380. //执行批量修改数据
  381. if (!empty($update)) {
  382. $sql = batchUpdate($rollbackdata, 'dest_login', ['orig_order' => $msg['orig_order']],"order_progress");
  383. $db->query($sql);
  384. }
  385. //成功执行删除后增加数据
  386. if ($delete_sql != 'delete from order_progress where ') {
  387. $delete_sql = trim($delete_sql, 'or ');
  388. $insert_sql = substr($insert_sql, 0, -1);
  389. $db->query($delete_sql);
  390. $db->query($insert_sql);
  391. if($login){
  392. $logins = substr($login,0,-1);
  393. balancefix($logins);
  394. }
  395. }
  396. }
  397. if ($msg['type'] == "except_rollback") { //retry表中的Rollback请求 (有数据库操作,插入并看看是否需要直接返回给前端)
  398. $recv_buffer = "";
  399. /* if (!isset($msg['seqnum'])) {
  400. return false;
  401. }
  402. if ($seqnum != $msg['seqnum']) {
  403. return false;
  404. }*/
  405. /* ack($msg["seqnum"]);
  406. $seqnum++;*/
  407. $retry_rollback_index++; //将数据加一
  408. $total = count($retry_rollback_data); //总数据的长度
  409. if ($retry_rollback_index < $total) {
  410. send_retry_rollback($retry_rollback_data, $retry_rollback_index); //继续发送数据
  411. } else {
  412. $retry_rollback_index = 0; //对序号重置为0
  413. }
  414. $exceptrollbackdata = $msg['desc'];
  415. sendToWebServer($msg);
  416. $update = [];
  417. $delete_sql = 'delete from order_progress where ';
  418. $insert_sql = "insert into order_save(type,orig_order,orig_login,dest_login,dest_order,percentage,profit,error_code,addtime) values";
  419. foreach ($exceptrollbackdata as $key => $value) {
  420. if ($value['error_code'] == -2) {
  421. return false;
  422. } else {
  423. $time = time();
  424. if ($value['error_code'] == 0) {
  425. $delete_sql .= "(dest_login = $value[dest_login] and orig_order = $msg[orig_order]) or ";
  426. $insert_sql .= "('$msg[type]',$msg[orig_order],$msg[orig_login],$value[dest_login],$value[dest_order],$value[percentage],$value[profit],$value[error_code],$time),";
  427. } else {
  428. $exceptrollbackdata[$key]['type'] = $msg['type'];
  429. $exceptrollbackdata[$key]['orig_order'] = $msg['orig_order'];
  430. $exceptrollbackdata[$key]['orig_login'] = $msg['orig_login'];
  431. $exceptrollbackdata[$key]['addtime'] = time();
  432. array_push($update, $value);
  433. }
  434. }
  435. }
  436. if (!empty($update)){
  437. $sql = batchUpdate($exceptrollbackdata, 'dest_login', ['orig_order' => $msg['orig_order']],"order_progress");
  438. $db->query($sql);
  439. }
  440. if ($delete_sql != 'delete from order_progress where ') {
  441. $delete_sql = trim($delete_sql, 'or ');
  442. $insert_sql = substr($insert_sql, 0, -1);
  443. $db->query($delete_sql);
  444. $db->query($insert_sql);
  445. }
  446. }
  447. if ($msg['type'] == "retry") { //
  448. /* if (!isset($msg['seqnum'])) {
  449. return false;
  450. }
  451. if ($seqnum != $msg['seqnum']) {
  452. return false;
  453. }*/
  454. /* ack($msg["seqnum"]);
  455. $seqnum++;*/
  456. $retry_index++; //将数据加一
  457. $total = count($retry_data); //总数据的长度
  458. if ($retry_index < $total) {
  459. send_retry($retry_data, $retry_index);
  460. } else {
  461. $retry_index = 0;
  462. }
  463. $retrydata = $msg['desc'];
  464. sendToWebServer($msg);
  465. $succeUpdate = [];
  466. $update = [];
  467. $insert_sql = "insert into order_save(type,orig_order,orig_login,dest_login,dest_order,percentage,profit,error_code,addtime) values";
  468. foreach ($retrydata as $key => $value) {
  469. if ($value['error_code'] == -2) {
  470. return false;
  471. } else {
  472. $time = time();
  473. if ($value['error_code'] == 0) {
  474. $retrydata[$key]['type'] = $msg['type'];
  475. $retrydata[$key]['orig_order'] = $msg['orig_order'];
  476. $retrydata[$key]['orig_login'] = $msg['orig_login'];
  477. $retrydata[$key]['addtime'] = time();
  478. array_push($succeUpdate, $value);
  479. $insert_sql .= "('$msg[type]',$msg[orig_order],$msg[orig_login],$value[dest_login],$value[dest_order],$value[percentage],$value[profit],$value[error_code],$time),";
  480. } else {
  481. $retrydata[$key]['type'] = $msg['type'];
  482. $retrydata[$key]['orig_order'] = $msg['orig_order'];
  483. $retrydata[$key]['orig_login'] = $msg['orig_login'];
  484. $retrydata[$key]['addtime'] = time();
  485. array_push($update, $value);
  486. }
  487. }
  488. }
  489. if (!empty($update)) {
  490. $sql = batchUpdate($retrydata, 'dest_login', ['orig_order' => $msg['orig_order']], "order_progress");
  491. $db->query($sql);
  492. }
  493. if (!empty($succeUpdate)) {
  494. $sql = batchUpdate($retrydata, 'dest_login', ['orig_order' => $msg['orig_order']], "order_progress");
  495. $db->query($sql);
  496. $insert_sql = substr($insert_sql, 0, -1);
  497. $db->query($insert_sql);
  498. }
  499. }
  500. }
  501. }
  502. //发送ack给远程服务器
  503. function ack($seqnum){
  504. global $connection_to_tcp;
  505. $data = ["type"=>"ack","seqnum"=>$seqnum];
  506. $connection_to_tcp->send(json_encode($data) . "\n");
  507. }
  508. // 发送信息到webserver服务器
  509. function sendToWebServer($data)
  510. {
  511. global $con;
  512. log_file($data);
  513. $con->send(json_encode($data) . "\n");
  514. }
  515. // 发送信息到接口服务器
  516. function sendTo_tcp_Server($data){
  517. global $connection_to_tcp;
  518. global $seqnum;
  519. // $data['seqnum'] = $seqnum;
  520. log_file($data);
  521. echo "发送给接口的信息";
  522. var_dump($data);
  523. $connection_to_tcp->send(json_encode($data) . "\n");
  524. }
  525. // insert 数据
  526. function send_insert($data,$index=0,$order){
  527. $send['type'] = "insert_some";
  528. $send['orig_order'] = (int)$order; //订单号码
  529. $send['dest'] =[];
  530. $send_data = $data[$index]; //将分割的数据交给全局变量
  531. foreach ($send_data as $key => $value) {
  532. $send_child['dest_login'] = $value['LOGIN']; //登录的账号
  533. $send_child['percentage'] = $value['ladder']; //梯度
  534. array_push($send['dest'],$send_child);
  535. }
  536. sendTo_tcp_Server($send);
  537. }
  538. // rollback 数据
  539. function send_rollback($data,$index=0){
  540. global $db;
  541. $send['type'] = "rollback_some";
  542. $send['orig_order'] = (int)$data[$index][0]['orig_order']; //订单号码
  543. $send['orig_login'] = (int)$data[$index][0]['orig_login']; //登录者账号
  544. $send['desc'] =[];
  545. $send_child =[];
  546. $send_data = $data[$index];//将分割的数据交给全局变量
  547. foreach ($send_data as $key => $value) {
  548. $order = $db->query("SELECT `Order` FROM mt5_orders_history WHERE PositionID = {$value['dest_order']}");
  549. $deals = $db->query("select `Deal` FROM mt5_deals WHERE PositionID = {$value['dest_order']}");
  550. $send_child['dest_login'] = (int)$value['dest_login']; //登录的账号
  551. $send_child['dest_orders'] = array_column($order,"Order"); //订单号
  552. $send_child['dest_deals'] = array_column($deals,"Deal");
  553. $send_child['percentage'] = (int)$value['percentage']; //手数
  554. array_push($send['desc'],$send_child);
  555. }
  556. var_dump($send);
  557. // sendTo_tcp_Server($send);
  558. }
  559. // retry表中的rollback操作
  560. function send_retry_rollback($data,$index=0){
  561. $send['type'] = "except_rollback_some";
  562. $send['orig_order'] = (int)$data[$index][0]['orig_order']; //订单号码
  563. $send['orig_login'] = (int)$data[$index][0]['orig_login']; //登录者账号
  564. $send['desc'] =[];
  565. $send_child =[];
  566. $send_data = $data[$index]; //将分割的数据交给全局变量
  567. foreach ($send_data as $key => $value) {
  568. $send_child['dest_login'] = (int)$value['dest_login']; //登录的账号
  569. $send_child['dest_order'] = (int)$value['dest_order']; //订单号
  570. $send_child['percentage'] = (int)$value['percentage']; //手数
  571. $send_child['profit'] = (float)$value['profit']; //利润点
  572. $send_child['error_code'] = (int)$value['error_code']; //错误的号码
  573. array_push($send['desc'],$send_child);
  574. }
  575. echo "发送的数据";
  576. var_dump($send);
  577. sendTo_tcp_Server($send);
  578. }
  579. // retry表中的retry操作
  580. function send_retry($data,$index=0){
  581. $send['type'] = "retry_some";
  582. $send['orig_order'] = (int)$data[$index][0]['orig_order']; //订单号码
  583. $send['orig_login'] = (int)$data[$index][0]['orig_login']; //登录者账号
  584. $send['desc'] =[];
  585. $send_child =[];
  586. $send_data = $data[$index]; //将分割的数据交给全局变量
  587. foreach ($send_data as $key => $value) {
  588. $send_child['dest_login'] = (int)$value['dest_login']; //登录的账号
  589. $send_child['dest_order'] = (int)$value['dest_order']; //订单号
  590. $send_child['percentage'] = (int)$value['percentage']; //手数
  591. $send_child['profit'] = (float)$value['profit']; //利润点
  592. $send_child['error_code'] = (int)$value['error_code']; //错误的号码
  593. array_push($send['desc'],$send_child);
  594. }
  595. echo "发送的数据";
  596. var_dump($send);
  597. sendTo_tcp_Server($send);
  598. }
  599. //获取批量修改sql
  600. function batchUpdate($data,$field,$params=[],$tableName)
  601. {
  602. if (!is_array($data) || !$field || !is_array($params)) {
  603. return false;
  604. }
  605. $updates = parseUpdate($data, $field);
  606. $where = parseParams($params);
  607. $fields = array_column($data, $field);
  608. $fields = implode(',', array_map(function($value) {
  609. return "'".$value."'";
  610. }, $fields));
  611. $sql = sprintf("UPDATE `%s` SET %s WHERE `%s` IN (%s) %s", "$tableName", $updates, $field, $fields, $where);
  612. return $sql;
  613. }
  614. function parseUpdate($data,$field)
  615. {
  616. $sql = "";
  617. $keys = array_keys(current($data));
  618. foreach ($keys as $column){
  619. $sql .= sprintf(" `%s` = CASE `%s` \n",$column,$field);
  620. foreach ($data as $line){
  621. $sql .= sprintf("WHEN '%s' THEN '%s' \n",$line[$field],$line[$column]);
  622. }
  623. $sql .= "END,";
  624. }
  625. return rtrim($sql,",");
  626. }
  627. function parseParams($params)
  628. {
  629. $where = [];
  630. foreach ($params as $key=>$value){
  631. $where[] = sprintf(" `%s` = '%s' ",$key,$value);
  632. }
  633. return $where ? " AND " .implode('AND',$where) : '';
  634. }
  635. function getToken()
  636. {
  637. global $token;
  638. $params = ['request'=>'login','login'=>LOGIN,'password'=>PASSWORD];
  639. $data = http(BALANCE_URL,$params);
  640. $data = json_decode($data,true);
  641. $token = $data['token'];
  642. }
  643. function refreshToken()
  644. {
  645. global $token;
  646. $sendToken = $token;
  647. $params = ['request'=>'token_renewal','token'=>$sendToken];
  648. $data = http(BALANCE_URL,$params);
  649. $data = json_decode($data,true);
  650. $token = $data['token'];
  651. }
  652. function balancefix($logins)
  653. {
  654. global $token;
  655. $params = ['logins'=>$logins,'request'=>'balancefix','token'=>$token];
  656. $data = http(BALANCE_URL,$params);
  657. var_dump($data);
  658. }
  659. //发送curl
  660. function http($url, $params, $method = 'GET', $header = array(), $timeout = 5)
  661. {
  662. // POST 提交方式的传入 $set_params 必须是字符串形式
  663. $opts = array(
  664. CURLOPT_TIMEOUT => $timeout,
  665. CURLOPT_RETURNTRANSFER => 1,
  666. CURLOPT_SSL_VERIFYPEER => false,
  667. CURLOPT_SSL_VERIFYHOST => false,
  668. CURLOPT_HTTPHEADER => $header
  669. );
  670. /* 根据请求类型设置特定参数 */
  671. switch (strtoupper($method)) {
  672. case 'GET':
  673. $opts[CURLOPT_URL] = $url . '?' . http_build_query($params);
  674. break;
  675. case 'POST':
  676. $params = http_build_query($params);
  677. $opts[CURLOPT_URL] = $url;
  678. $opts[CURLOPT_POST] = 1;
  679. $opts[CURLOPT_POSTFIELDS] = $params;
  680. break;
  681. case 'DELETE':
  682. $opts[CURLOPT_URL] = $url;
  683. $opts[CURLOPT_HTTPHEADER] = array("X-HTTP-Method-Override: DELETE");
  684. $opts[CURLOPT_CUSTOMREQUEST] = 'DELETE';
  685. $opts[CURLOPT_POSTFIELDS] = $params;
  686. break;
  687. case 'PUT':
  688. $opts[CURLOPT_URL] = $url;
  689. $opts[CURLOPT_POST] = 0;
  690. $opts[CURLOPT_CUSTOMREQUEST] = 'PUT';
  691. $opts[CURLOPT_POSTFIELDS] = $params;
  692. break;
  693. default:
  694. throw new Exception('不支持的请求方式!');
  695. }
  696. /* 初始化并执行curl请求 */
  697. $ch = curl_init();
  698. curl_setopt_array($ch, $opts);
  699. $data = curl_exec($ch);
  700. $error = curl_error($ch);
  701. return $data;
  702. }
  703. //程序执行时间秒速
  704. /*function curenttime()
  705. {
  706. $t = microtime(true);
  707. $micro = sprintf("%06d",($t - floor($t)) * 1000000);
  708. $d = new DateTime( date('Y-m-d H:i:s.'.$micro, $t) );
  709. print $d->format("Y-m-d H:i:s.u"); // note at point on "u"
  710. }*/
  711. //自定义日志类
  712. function log_file($data){
  713. $data = print_r($data,true);
  714. error_log(date("Y-m-d H:i:s", time()).' ===== info: '.$data."\n",3,__DIR__ . '/log/' . date("Ymd", time()) . '.log');
  715. }
  716. Worker::runAll(); // 执行函数