|
|
@@ -20,15 +20,18 @@ date_default_timezone_set("PRC");
|
|
|
|
|
|
//初始化 (创建了一个tcp服务)
|
|
|
|
|
|
-$workerTcp = new Worker("tcp://0.0.0.0:12346");
|
|
|
+$workerTcp = new Worker("tcp://0.0.0.0:12347");
|
|
|
|
|
|
$workerTcp->name = "TCP";
|
|
|
-
|
|
|
$workerTcp::$logFile = __DIR__.'/workerman.log';
|
|
|
+$path = "displaylog/".date('Ym', time());
|
|
|
+createdir($path);
|
|
|
+$workerTcp::$stdoutFile = __DIR__."/".$path."/".date("Ymd",time()).'.log';
|
|
|
+
|
|
|
$num = 1;
|
|
|
|
|
|
$workerTcp->onWorkerStart = function ($workerTcp) {
|
|
|
- global $dbMt5;
|
|
|
+
|
|
|
global $db; //数据库
|
|
|
global $con; //访问本地创建的webserver服务器
|
|
|
global $connection_to_tcp; //访问接口服务器(远程的)
|
|
|
@@ -51,10 +54,9 @@ $workerTcp->onWorkerStart = function ($workerTcp) {
|
|
|
$seqnum = 1;
|
|
|
getToken();
|
|
|
$db = new Connection(DB_HOST, DB_PORT, DB_USERNAME, DB_PASSWORD, DB_NAME); //本地数据库的配置
|
|
|
- $dbMt5 = new Connection("43.249.29.218","23891","mt5devel","Pn9OZuvfmJFGZf4qsDQuYA32q3ryVPV0","mt5backup");
|
|
|
|
|
|
//访问本地的websocket服务器=============================================
|
|
|
- $con = new AsyncTcpConnection("ws://127.0.0.1:12381");
|
|
|
+ $con = new AsyncTcpConnection("ws://127.0.0.1:12382");
|
|
|
$con->onConnect = function ($on) {
|
|
|
global $index;
|
|
|
global $rollback_index;
|
|
|
@@ -115,7 +117,7 @@ $workerTcp->onWorkerStart = function ($workerTcp) {
|
|
|
|
|
|
global $rollback_index; //传递index
|
|
|
global $rollback_data; //全局的数据
|
|
|
- $rollback_data = array_chunk($msg['data'], 5); //每几个是一个数组
|
|
|
+ $rollback_data = array_chunk($msg['data'], 8); //每几个是一个数组
|
|
|
send_rollback($rollback_data, $rollback_index);
|
|
|
|
|
|
}
|
|
|
@@ -138,7 +140,7 @@ $workerTcp->onWorkerStart = function ($workerTcp) {
|
|
|
};
|
|
|
$con->onClose = function ($con) {
|
|
|
echo "connection closed\n";
|
|
|
- $con->reConnect(1);
|
|
|
+ //$con->reConnect(1);
|
|
|
};
|
|
|
|
|
|
$con->onError = function ($con, $code, $msg) {
|
|
|
@@ -175,7 +177,7 @@ $workerTcp->onWorkerStart = function ($workerTcp) {
|
|
|
echo "connection closed\n";
|
|
|
log_file("to_tcp connection closed\n");
|
|
|
|
|
|
- // $connection_to_tcp->reConnect(); //执行连接
|
|
|
+ $connection_to_tcp->reConnect(); //执行连接
|
|
|
};
|
|
|
|
|
|
$connection_to_tcp->onError = function($connection_to_tcp, $code, $msg)
|
|
|
@@ -204,10 +206,10 @@ $workerTcp->onWorkerStart = function ($workerTcp) {
|
|
|
// 定时发送心跳信息
|
|
|
global $connection_count;
|
|
|
|
|
|
- if($connection_count){
|
|
|
+ /*if($connection_count){
|
|
|
if($connection_count>1){
|
|
|
|
|
|
- $connection_to_tcp->connect(); //客户端上线重新连接接口
|
|
|
+
|
|
|
$connection_to_tcp->onConnect = function ($connection_to_tcp) {
|
|
|
// 发送心跳信息
|
|
|
$connection_to_tcp->send('{"type":"ping"}' . "\n");
|
|
|
@@ -215,13 +217,13 @@ $workerTcp->onWorkerStart = function ($workerTcp) {
|
|
|
// 接口服务器向我发送的信息
|
|
|
$connection_to_tcp->onMessage = function ($connection_to_tcp, $data) {
|
|
|
|
|
|
- tcpMessHandle($data);
|
|
|
+ tcpMessHandle($data);
|
|
|
};
|
|
|
|
|
|
$connection_to_tcp->onClose = function($connection_to_tcp)
|
|
|
{
|
|
|
|
|
|
- echo "connection closed\n";
|
|
|
+ echo "connection closed\n";
|
|
|
log_file("to_tcp connection closed\n");
|
|
|
|
|
|
// $connection_to_tcp->reConnect(); //执行连接
|
|
|
@@ -231,24 +233,24 @@ $workerTcp->onWorkerStart = function ($workerTcp) {
|
|
|
{
|
|
|
echo "Error code:$code msg:$msg\n";
|
|
|
};
|
|
|
+ $connection_to_tcp->connect(); //客户端上线重新连接接口
|
|
|
|
|
|
}else{
|
|
|
//客户端不在线
|
|
|
- $connection_to_tcp->close();
|
|
|
+ $connection_to_tcp->close();
|
|
|
}
|
|
|
- }
|
|
|
+ }*/
|
|
|
$con->send(json_encode(['type' => 'heartbeat']). "\n");
|
|
|
});
|
|
|
|
|
|
|
|
|
- /*Timer::add(,function(){
|
|
|
- global $token;
|
|
|
- refreshToken();
|
|
|
+ Timer::add(1800,function(){
|
|
|
|
|
|
+ getToken();
|
|
|
|
|
|
- });*/
|
|
|
+ });
|
|
|
// 心跳检测(TCP 所有客户端) 心跳===================================================
|
|
|
- Timer::add(30, function () use ($workerTcp) {
|
|
|
+ Timer::add(1, function () use ($workerTcp) {
|
|
|
$time_now = time(); //当前时间
|
|
|
foreach ($workerTcp->connections as $connection) {
|
|
|
// 有可能该connection还没收到过消息，则lastMessageTime设置为当前时间
|
|
|
@@ -264,6 +266,11 @@ $workerTcp->onWorkerStart = function ($workerTcp) {
|
|
|
}
|
|
|
});
|
|
|
|
|
|
+ //避免数据库长时间未操作连接超时
|
|
|
+ Timer::add(25200,function(){
|
|
|
+ global $db;
|
|
|
+ $db->query("select username from admin limit 1");
|
|
|
+ });
|
|
|
};
|
|
|
|
|
|
$workerTcp->onConnect = function ($connection) {
|
|
|
@@ -327,8 +334,7 @@ function tcpMessHandle($data)
|
|
|
}
|
|
|
|
|
|
// 接口服务器返回的信息
|
|
|
- echo "接口返回数据";
|
|
|
- var_dump($msg);
|
|
|
+
|
|
|
if ($msg['type'] == "get_order_record") { //查询订单状态
|
|
|
$recv_buffer = "";
|
|
|
/* if (!isset($msg['seqnum'])) {
|
|
|
@@ -350,6 +356,9 @@ function tcpMessHandle($data)
|
|
|
|
|
|
if ($msg['type'] == "insert") { //Insert请求 (有数据库操作,插入)
|
|
|
$recv_buffer = "";
|
|
|
+ echo '接口返回数据';
|
|
|
+ curenttime();
|
|
|
+ var_dump($msg);
|
|
|
/* if (!isset($msg['seqnum'])) {
|
|
|
return false;
|
|
|
}
|
|
|
@@ -374,6 +383,11 @@ function tcpMessHandle($data)
|
|
|
$arr = "";
|
|
|
$time = time();
|
|
|
$delete_sql = 'delete from order_progress where ';
|
|
|
+ foreach($insert_data as $key=>$value){
|
|
|
+ $insert_data[$key]['dest_orders'] =json_encode($value['dest_orders'],JSON_FORCE_OBJECT);
|
|
|
+ $insert_data[$key]['dest_deals'] = json_encode($value['dest_deals'],JSON_FORCE_OBJECT);
|
|
|
+ }
|
|
|
+
|
|
|
|
|
|
foreach ($insert_data as $key => $value) {
|
|
|
|
|
|
@@ -382,13 +396,13 @@ function tcpMessHandle($data)
|
|
|
} else {
|
|
|
if($message == 1){
|
|
|
if($value['error_code'] == 0){
|
|
|
- $arr .= "(" . "'{$msg['type']}'" . "," . $msg['orig_position'] . "," . $msg['orig_login'] . "," . $value['dest_login'] . "," . $value['dest_position'] . "," . $value['percentage'] . "," . $value['profit'] . "," . $value['error_code'] . "," . $time . ")" . ",";
|
|
|
+ $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']."'".")" . ",";
|
|
|
$delete_sql .= "(dest_login = $value[dest_login] and orig_order = $msg[orig_position] and error_code != 0) or ";
|
|
|
$login .= $value['dest_login'].",";
|
|
|
}
|
|
|
}else{
|
|
|
|
|
|
- $arr .= "(" . "'{$msg['type']}'" . "," . $msg['orig_position'] . "," . $msg['orig_login'] . "," . $value['dest_login'] . "," . $value['dest_position'] . "," . $value['percentage'] . "," . $value['profit'] . "," . $value['error_code'] . "," . $time . ")" . ",";
|
|
|
+ $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']."'". ")" . ",";
|
|
|
if($value['error_code'] == 0){
|
|
|
|
|
|
$login .= $value['dest_login'].",";
|
|
|
@@ -401,22 +415,25 @@ function tcpMessHandle($data)
|
|
|
}
|
|
|
|
|
|
|
|
|
- $filed = "type,orig_order,orig_login,dest_login,dest_order,percentage,profit,error_code,addtime";
|
|
|
+ $filed = "type,orig_order,orig_login,dest_login,dest_order,percentage,profit,error_code,addtime,dest_orders,dest_deals";
|
|
|
$arr = substr($arr, 0, -1);
|
|
|
$sql = sprintf("INSERT INTO %s(%s) VALUES %s ", "order_progress", $filed, $arr);
|
|
|
- $sql2 = sprintf("INSERT INTO %s(%s) VALUES %s", "order_save", $filed, $arr);
|
|
|
+
|
|
|
+ //$sql2 = sprintf("INSERT INTO %s(%s) VALUES %s", "order_save", $filed, $arr);
|
|
|
|
|
|
if($message == 1){
|
|
|
if ($delete_sql != 'delete from order_progress where ') {
|
|
|
$delete_sql = trim($delete_sql, 'or ');
|
|
|
$db->query($delete_sql);
|
|
|
$db->query($sql);
|
|
|
- $db->query($sql2);
|
|
|
+ //$db->query($sql2);
|
|
|
}
|
|
|
|
|
|
}else{
|
|
|
- $db->query($sql);
|
|
|
- $db->query($sql2);
|
|
|
+ $data = $db->query($sql);
|
|
|
+
|
|
|
+
|
|
|
+ // $db->query($sql2);
|
|
|
}
|
|
|
|
|
|
|
|
|
@@ -434,6 +451,8 @@ function tcpMessHandle($data)
|
|
|
sendToWebServer($msg); //发送给前端
|
|
|
}
|
|
|
if ($msg['type'] == "rollback") { //Rollback请求 (有数据库操作,插入并看看是否需要直接返回给前端)
|
|
|
+ echo '接口返回数据'.curenttime();
|
|
|
+ var_dump($msg);
|
|
|
$recv_buffer = "";
|
|
|
/* if (!isset($msg['seqnum'])) {
|
|
|
return false;
|
|
|
@@ -455,14 +474,14 @@ function tcpMessHandle($data)
|
|
|
|
|
|
sendToWebServer($msg); //发送给前端
|
|
|
$delete_sql = 'delete from order_progress where ';
|
|
|
- $insert_sql = "insert into order_save(type,orig_order,orig_login,dest_login,dest_orders,dest_deals,percentage,error_code,addtime) values";
|
|
|
+ /* $insert_sql = "insert into order_save(type,orig_order,orig_login,dest_login,dest_orders,dest_deals,percentage,error_code,addtime) values";*/
|
|
|
|
|
|
$update = [];
|
|
|
$login = '';
|
|
|
- foreach($rollbackdata as $key=>$value){
|
|
|
- $rollbackdata[$key]['dest_orders'] = json_encode($rollbackdata[$key]['dest_orders'],JSON_FORCE_OBJECT);
|
|
|
- $rollbackdata[$key]['dest_deals'] = json_encode($rollbackdata[$key]['dest_orders'],JSON_FORCE_OBJECT);
|
|
|
- }
|
|
|
+ /* foreach($rollbackdata as $key=>$value){
|
|
|
+ $rollbackdata[$key]['dest_orders'] = json_encode($rollbackdata[$key]['dest_orders'],JSON_FORCE_OBJECT);
|
|
|
+ $rollbackdata[$key]['dest_deals'] = json_encode($rollbackdata[$key]['dest_orders'],JSON_FORCE_OBJECT);
|
|
|
+ }*/
|
|
|
|
|
|
foreach ($rollbackdata as $key => $value) {
|
|
|
|
|
|
@@ -474,8 +493,8 @@ function tcpMessHandle($data)
|
|
|
if ($value['error_code'] == 0) {
|
|
|
|
|
|
$delete_sql .= "(dest_login = $value[dest_login] and orig_order = $msg[orig_position]) or ";
|
|
|
- var_dump($value['dest_login']);
|
|
|
- $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),";
|
|
|
+
|
|
|
+ /*$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),";*/
|
|
|
$login .= $value['dest_login'].",";
|
|
|
} else {
|
|
|
//不成功把参数组装数组以便批量修改
|
|
|
@@ -501,18 +520,18 @@ function tcpMessHandle($data)
|
|
|
//成功执行删除后增加数据
|
|
|
if ($delete_sql != 'delete from order_progress where ') {
|
|
|
$delete_sql = trim($delete_sql, 'or ');
|
|
|
- $insert_sql = substr($insert_sql, 0, -1);
|
|
|
+ //$insert_sql = substr($insert_sql, 0, -1);
|
|
|
$db->query($delete_sql);
|
|
|
- $db->query($insert_sql);
|
|
|
+ // $db->query($insert_sql);
|
|
|
+
|
|
|
if($login){
|
|
|
|
|
|
$logins = substr($login,0,-1);
|
|
|
- var_dump($logins);
|
|
|
+
|
|
|
balancefix($logins);
|
|
|
|
|
|
}
|
|
|
|
|
|
-
|
|
|
}
|
|
|
|
|
|
}
|
|
|
@@ -660,7 +679,7 @@ function sendTo_tcp_Server($data){
|
|
|
global $seqnum;
|
|
|
// $data['seqnum'] = $seqnum;
|
|
|
log_file($data);
|
|
|
- echo "发送给接口的信息";
|
|
|
+ echo "发送给接口的信息".curenttime();
|
|
|
var_dump($data);
|
|
|
$connection_to_tcp->send(json_encode($data) . "\n");
|
|
|
}
|
|
|
@@ -684,19 +703,19 @@ function send_insert($data,$index=0,$order){
|
|
|
// rollback 数据
|
|
|
function send_rollback($data,$index=0){
|
|
|
|
|
|
- global $dbMt5;
|
|
|
+ global $db;
|
|
|
$send['type'] = "rollback_some";
|
|
|
$send['orig_position'] = (int)$data[$index][0]['orig_order']; //订单号码
|
|
|
$send['orig_login'] = (int)$data[$index][0]['orig_login']; //登录者账号
|
|
|
$send['desc'] =[];
|
|
|
$send_child =[];
|
|
|
$send_data = $data[$index]; //将分割的数据交给全局变量
|
|
|
+
|
|
|
foreach ($send_data as $key => $value) {
|
|
|
- $order = $dbMt5->query("SELECT `Order` FROM mt5_orders_history WHERE PositionID = {$value['dest_order']}");
|
|
|
- $deals = $dbMt5->query("SELECT `Deal` FROM mt5_deals WHERE PositionID = {$value['dest_order']}");
|
|
|
+ //$order = $db->query("SELECT `Order`,`Deal` FROM mt5_deals WHERE PositionID = {$value['dest_order']}");
|
|
|
$send_child['dest_login'] = (int)$value['dest_login']; //登录的账号
|
|
|
- $send_child['dest_orders'] = array_column($order,'Order'); //订单号
|
|
|
- $send_child['dest_deals'] = array_column($deals,'Deal');
|
|
|
+ $send_child['dest_orders'] = json_decode($value['dest_orders'],true); //订单号
|
|
|
+ $send_child['dest_deals'] = json_decode($value['dest_deals'],true);
|
|
|
$send_child['percentage'] = (int)$value['percentage'];
|
|
|
array_push($send['desc'],$send_child);
|
|
|
}
|
|
|
@@ -823,8 +842,11 @@ function balancefix($logins)
|
|
|
$params = ['logins'=>$logins,'token'=>$token];
|
|
|
$data = http(BALANCE_URL.$requestApi,$params,"POST");
|
|
|
$data = json_decode($data,true);
|
|
|
+ echo 'balancefix';
|
|
|
+ curenttime();
|
|
|
var_dump($data);
|
|
|
if(!$data){
|
|
|
+ log_file("send balancefix Invalid\n");
|
|
|
getToken();
|
|
|
balancefix($logins);
|
|
|
}
|
|
|
@@ -879,13 +901,13 @@ function http($url, $params, $method = 'GET', $header = array(), $timeout = 5)
|
|
|
}
|
|
|
|
|
|
//程序执行时间秒速
|
|
|
-/*function curenttime()
|
|
|
+function curenttime()
|
|
|
{
|
|
|
$t = microtime(true);
|
|
|
$micro = sprintf("%06d",($t - floor($t)) * 1000000);
|
|
|
$d = new DateTime( date('Y-m-d H:i:s.'.$micro, $t) );
|
|
|
print $d->format("Y-m-d H:i:s.u"); // note at point on "u"
|
|
|
-}*/
|
|
|
+}
|
|
|
|
|
|
|
|
|
|
|
|
@@ -917,5 +939,23 @@ function log_file($data){
|
|
|
error_log(date("Y-m-d H:i:s", time()).' ===== info: '.$data."\n",3,__DIR__ . '/log/' . date("Ymd", time()) . '.log');
|
|
|
}
|
|
|
|
|
|
+function createdir($path) {
|
|
|
+ // 判断传过来的$path是否已是目录，若是，则直接返回true
|
|
|
+ if(is_dir($path)) {
|
|
|
+ return true;
|
|
|
+ }
|
|
|
+
|
|
|
+ // 走到这步，说明传过来的$path不是目录
|
|
|
+ // 判断其上级是否为目录，是，则直接创建$path目录
|
|
|
+ if(is_dir(dirname($path))) {
|
|
|
+ return mkdir($path);
|
|
|
+ }
|
|
|
|
|
|
+ // 走到这说明其上级目录也不是目录,则继续判断其上上...级目录
|
|
|
+ createdir(dirname($path));
|
|
|
+
|
|
|
+ // 走到这步，说明上级目录已创建成功，则直接接着创建当前目录，并把创建的结果返回
|
|
|
+ return mkdir($path);
|
|
|
+
|
|
|
+}
|
|
|
Worker::runAll(); // 执行函数
|
