tcpServer.php 29 KB

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