kshxt/thinkphp/TCPserver/tcpServer.php

<?php


require dirname(dirname(__FILE__)).'/vendor/autoload.php';
require_once dirname(dirname(__FILE__)) . '/vendor/workerman/Autoloader.php';
require_once dirname(dirname(__FILE__)) . '/vendor/mysql/src/Connection.php';
require_once __DIR__ . '/config.php';


use Workerman\Worker;                             //worker容器类
use Workerman\Connection\AsyncTcpConnection;      //连接异步客户端
use Workerman\MySQL\Connection;                   //连接数据库的类
use Workerman\Lib\Timer;                          //定时器函数

use Monolog\Logger;
use Monolog\Handler\StreamHandler;
use Monolog\Handler\ErrorLogHandler;



//初始化  (创建了一个tcp服务)

$workerTcp = new Worker("tcp://127.0.0.1:12345");

$workerTcp->name = "TCP";

$workerTcp::$logFile = __DIR__.'/workerman.log';


$workerTcp->onWorkerStart = function ($workerTcp) {

        global $db;                                  //数据库
        global $con;                                 //访问本地创建的webserver服务器
        global $connection_to_tcp;                   //访问接口服务器(远程的)

        global $send_data;       //存储insert数据
        global $index;           //insert管理数据序号
        global $send_order;      //insert全局的订单号


        global $rollback_data;   //rollback数据
        global $rollback_index;  //rollback 管理数据序号


        $db = new Connection(DB_HOST, DB_PORT, DB_USERNAME, DB_PASSWORD, DB_NAME);  //本地数据库的配置


        //访问本地的websocket服务器=============================================
        $con = new AsyncTcpConnection("ws://127.0.0.1:12380");    
        $con->onConnect = function ($on) {
            global $index;
            global $rollback_index;
            $index=0;
            $rollback_index=0;
            sendToWebServer([
                'type' => 'login',
                'uid' => 'tcp'
            ]);
        };

        // websocke服务器发送信息过来触发的函数
        $con->onMessage = function ($con, $data) {
            global $send_data;
            $msg = json_decode($data, true);
            if ($msg['type'] == "get_order_record") {      //查询订单状态
                sendTo_tcp_Server($msg);
            }
            if ($msg['type'] == "set_current_order") {     //设置当前跟单订单
                sendTo_tcp_Server($msg);
            }
            if ($msg['type'] == "insert") {                //Insert请求

                global $index;              //传递的index
                global $send_data;          //全局的数据
                global $send_order;         //全局的订单号码
                $insert_data = $msg['data'];
                $insert_data = array_chunk($insert_data,12);      //每十个是一个数组
                $send_data = $insert_data;   //传递的数据
                $send_order = (int)$msg['orderid'];
                send_insert($send_data,$index,$send_order);

            }
            if ($msg['type'] == "rollback") {                      //Rollback请求 

                global $rollback_index;  //传递index
                global $rollback_data;   //全局的数据
                $rollback_data = array_chunk($msg['data'],5);      //每几个是一个数组
                send_rollback($rollback_data,$rollback_index);

            }

            if ($msg['type'] == "except_rollback") {              //retry表中的 Rollback请求 
                $orllback_data = $msg['data'];
                foreach ($orllback_data as $key => $value) {
                    $send['type'] = $msg['type'];                      //rollback类型
                    $send['orig_order'] = (int)$value['orig_order'];   //手数
                    $send['orig_login'] = (int)$value['orig_login'];   //订单号码
                    $send['dest_login'] = (int)$value['dest_login'];   //登录的账号
                    $send['dest_order'] = (int)$value['dest_order'];   //登录的账号
                    $send['error_code'] = (int)$value['error_code'];   //登录的账号
                    $send['percentage'] = (int)$value['percentage'];   //手数
                    $send['profit'] = (float)$value['profit'];         //利润点
                    sendTo_tcp_Server($send);
                }
                
            }


            if ($msg['type'] == "retry") {                //Retry请求
                $orllback_data = $msg['data'];
                foreach ($orllback_data as $key => $value) {
                    $send['type'] = $msg['type'];                      //rollback类型
                    $send['orig_order'] = (int)$value['orig_order'];   //手数
                    $send['orig_login'] = (int)$value['orig_login'];   //订单号码
                    $send['dest_login'] = (int)$value['dest_login'];   //登录的账号
                    $send['dest_order'] = (int)$value['dest_order'];   //登录的账号
                    $send['error_code'] = (int)$value['error_code'];   //登录的账号
                    $send['percentage'] = (int)$value['percentage'];   //手数
                    $send['profit'] = (float)$value['profit'];           //利润点
                    sendTo_tcp_Server($send);
                }
            }
        };
        $con->onClose = function ($con) {
            echo "connection closed\n";
            $con->reConnect(1);
        };

        $con->onError = function($con, $code, $msg)
        {
            echo "Error code:$code msg:$msg\n";  
        };
        
        $con->connect();

        // 访问本地的websocket服务器=======================================================
 


        // 向远处的服务器发送信息===========================================================
        $connection_to_tcp = new AsyncTcpConnection('tcp://47.254.202.24:10009');  
        // $connection_to_tcp = new AsyncTcpConnection('tcp://127.0.0.1:1235');  

        $connection_to_tcp->onConnect = function($connection_to_tcp)
        {
            // 发送心跳信息
            $connection_to_tcp->send('{"type":"ping"}' . "\r\n");
        };


        // 接口服务器向我发送的信息
        $connection_to_tcp->onMessage = function($connection_to_tcp, $data)
        {
            //这里显示的是从接口服务器返回的数据(需要根据数据的类似判断)   
            global $db;   //操作数据的全局变量

            global $index;          //全局的变量
            global $send_data;      //全局的数据
            global $send_order;     //全局的订单号码

            global $rollback_data;   //rollback数据
            global $rollback_index;  //rollback 管理数据序号

            $bufExplode = explode("\r\n", $data); //通过指定的分隔符,把字符串打散为数组
            foreach ($bufExplode as $key => $value) {
                
                $msg = json_decode($value, true);  //对信息进行处理
                   // 接口服务器返回的信息
                if ($msg['type'] == "get_order_record") {      //查询订单状态
                    
                    sendToWebServer($msg);

                }
                if ($msg['type'] == "set_current_order") {     //设置当前跟单订单

                    sendToWebServer($msg);
                    
                }

                if ($msg['type'] == "insert") {                //Insert请求 (有数据库操作,插入)
                    $index++;  //将数据加一
                    $total = count($send_data);  //总数据的长度
                    if($index < $total){
                        send_insert($send_data,$index,$send_order); 
                    }else{
                        $index = 0;
                    }
                    $insert_data = $msg['dest'];
                    sendToWebServer($msg); //发送给前端
                    // $insert_data_length = count($insert_data);
                    // $insert_data_index =0;
                    foreach ($insert_data as $key => $value) {
                            if($value['error_code']==-2){
                                return false;
                            }else{
                                $msg_child['type'] =       $msg['type'];
                                $msg_child['orig_order'] = $msg['orig_order'];
                                $msg_child['orig_login'] = $msg['orig_login'];
                                $msg_child['dest_login'] = $value['dest_login'];
                                $msg_child['dest_order'] = $value['dest_order'];
                                $msg_child['percentage'] = $value['percentage'];
                                $msg_child['profit']     = $value['profit'];
                                $msg_child['error_code'] = $value['error_code'];
                                $msg_child['addtime']=time();       //添加时间
                                $result = $db->insert('order_progress')->cols($msg_child)->query();             //插入数据
                                //插入成功,向前台发起
                                if($result){
                                    // $insert_data_index++;
                                }
                                
                            }
                    }

                }
                if ($msg['type'] == "rollback") {              //Rollback请求 (有数据库操作,插入并看看是否需要直接返回给前端) 
                    $rollback_index++;  //将数据加一
                    $total = count($rollback_data);  //总数据的长度
                    if($rollback_index < $total){
                        send_rollback($rollback_data,$rollback_index); 
                    }else{
                        $rollback_index = 0;
                    }
                    $rollbackdata = $msg['desc'];
                    sendToWebServer($msg);       //发送给前端

                    foreach ($rollbackdata as $key => $value) {
                        if($value['error_code']==-2){
                            return false;
                        }else{
                            $msg_child['type'] =       $msg['type'];
                            $msg_child['orig_order'] = $msg['orig_order'];
                            $msg_child['orig_login'] = $msg['orig_login'];
                            $msg_child['dest_login'] = $value['dest_login'];
                            $msg_child['dest_order'] = $value['dest_order'];
                            $msg_child['percentage'] = $value['percentage'];
                            $msg_child['profit']     = $value['profit'];
                            $msg_child['error_code'] = $value['error_code'];
                            $msg_child['addtime']=time();         //添加时间 
                            $dest_login = $value['dest_login'];   //获取更新的用户号码
                            $orig_order = $msg['orig_order'];
                            if($value['error_code']==0){
                            $result =  $db->delete('order_progress')->where('dest_login = :dest_login AND  orig_order= :orig_order')->bindValues(
                                array("dest_login" => $value['dest_login'],"orig_order" =>$msg['orig_order'])
                                )->query();//成功rollback 对其中的数据进行删除
                            }else{
                            $result = $db->update('order_progress')->cols($msg_child)->where(["dest_login = $dest_login AND orig_order = $orig_order"])->query();   //不成功就更新数据
                            }
        
                        } 
                    }
                }

                if ($msg['type'] == "except_rollback") {       //retry表中的Rollback请求 (有数据库操作,插入并看看是否需要直接返回给前端) 
                    if($msg['error_code']==-2){
                    log_file($msg['error_code']);
                    }else{

                        $msg['addtime'] = time();                 //更新时间
                        $dest_login = $msg['dest_login'];         //获取更新的用户号码
                        $orig_order = $msg['orig_order']; 
                        if($msg['error_code']==0){
                            $result =  $db->delete('order_progress')->where('dest_login = :dest_login AND  orig_order= :orig_order')->bindValues(
                                array("dest_login" => $value['dest_login'],"orig_order" =>$msg['orig_order'])
                                )->query();//成功rollback 对其中的数据进行删除
                            if($result){
                                sendToWebServer($msg); 
                            }
                        }else{
                            $result = $db->update('order_progress')->cols($msg)->where(["dest_login = $dest_login AND orig_order = $orig_order"])->query();   //不成功就更新数据
                        if($result){
                            sendToWebServer($msg);
                        }
                        }

                    }
                }


                if ($msg['type'] == "retry") {                //Retry请求 (有数据库操作 0成功然后更新数据库 )
                    if($msg['error_code']==-2){
                        return false;
                    }else{
                        
                        $msg['addtime']=time();                  //更新时间
                        $dest_login = $msg['dest_login'];        //获取更新的用户号码 
                        $orig_order = $msg['orig_order'];
                        $result = $db->update('order_progress')->cols($msg)->where(["dest_login = $dest_login AND orig_order = $orig_order"])->query();   //不成功就更新数据
                        // 更新成功,发送请求
                        if($result){
                            sendToWebServer($msg);
                        }

                    }
                }
            }
           
            
           


        };


        $connection_to_tcp->onClose = function($connection_to_tcp)
        {

            echo "connection closed\n";
            log_file("to_tcp connection closed\n");
            $connection_to_tcp->reConnect(1);  //重新连接

        };

        $connection_to_tcp->onError = function($connection_to_tcp, $code, $msg)
        {
            echo "Error code:$code msg:$msg\n";  
        };

        $connection_to_tcp->connect();  //执行连接

        // 向远处的服务器发送信息============================================================



        // 发送给 webserver 服务器  心跳=====================================================
        Timer::add(30, function () use ($con) {

            // 定时发送心跳信息
            $con->send(json_encode([
                'type' => 'heartbeat'
            ]));
        });


        //发送给  接口tcp服务器     心跳========================================================
        Timer::add(30, function () use ($connection_to_tcp) {

            //定时发送心跳信息
            $connection_to_tcp->send('{"type":"ping"}' . "\r\n");

        });

        // 心跳检测(TCP 所有客户端) 心跳===================================================
        Timer::add(1, function () use ($workerTcp) {
            $time_now = time();      //当前时间
            foreach ($workerTcp->connections as $connection) {
                // 有可能该connection还没收到过消息，则lastMessageTime设置为当前时间
                if (empty($connection->lastMessageTime)) {
                    $connection->lastMessageTime = $time_now;
                    continue;
                }
                // 上次通讯时间间隔大于心跳间隔，则认为客户端已经下线，关闭连接
                if ($time_now - $connection->lastMessageTime > HEARTBEAT_TIME) {

                    $connection->close();
                }
            } 
        });

};

$workerTcp->onConnect = function ($connection) {
    echo "connection onConnect\n";
};

$workerTcp->onError = function ($connection, $code, $msg) {
    echo "Error code:$code msg:$msg\n";
};

$workerTcp->onClose = function ($connection) {
    echo "connection closed\n";
};

$workerTcp->onMessage = function ($connection, $data) {
    echo "connection onMessage\n";
};


// 发送信息到webserver服务器
function sendToWebServer($data)
{
    global $con;
    $con->send(json_encode($data) . "\r\n");
}


// 发送信息到接口服务器
function sendTo_tcp_Server($data){

    global $connection_to_tcp;
    $connection_to_tcp->send(json_encode($data) . "\r\n");

}

// insert 数据
function send_insert($data,$index=0,$order){
    
    $send['type'] = "insert_some";
    $send['orig_order'] = (int)$order;     //订单号码
    $send['dest'] =[];
    $send_data = $data[$index];                                //将分割的数据交给全局变量
    foreach ($send_data as $key => $value) {
        $send_child['dest_login'] = $value['LOGIN'];           //登录的账号
        $send_child['percentage'] = intval($value['BALANCE']/$value['step']);    //手数
        array_push($send['dest'],$send_child);
    }
    echo "发送的数据";
    var_dump($send);
    sendTo_tcp_Server($send); 
}

// rollback 数据
function send_rollback($data,$index=0){

    $send['type'] = "rollback_some";
    $send['orig_order'] = (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) {
        $send_child['dest_login'] = (int)$value['dest_login'];   //登录的账号
        $send_child['dest_order'] = (int)$value['dest_order'];   //订单号
        $send_child['percentage'] = (int)$value['percentage'];   //手数
        $send_child['profit'] =     (float)$value['profit'];     //利润点
        $send_child['error_code'] = (int)$value['error_code'];   //错误的号码
        array_push($send['desc'],$send_child);
    }
    echo "发送的数据";
    var_dump($send);
    sendTo_tcp_Server($send); 
}

//自定义日志类
function log_file($data){
    error_log(date("Y-m-d H:i:s", time()).' ===== info: '.$data."\r\n",3,__DIR__ . '/log/' . date("Ymd", time()) . '.log');
}


Worker::runAll();   // 执行函数