开发更新

websockServ/app/lib/DB_pool.php

@@ -41,7 +41,7 @@ class DB_pool
         $this->worker_num = 5;
         $this->task_worker_num = 5;
         $this->dispatch_mode = 2;
-        $this->daemonize = 0;
+        $this->daemonize = 1;
         $this->max_request = 10000;
         $filename = date("Y-m-d", time());
         $this->log_file = "../logs/mysqlpoole_" . $filename . '.log';

websockServ/app/lib/Mconsts.php

@@ -22,13 +22,15 @@ defined('RUN_LOGS_QUEUE_KEY') or define("RUN_LOGS_QUEUE_KEY", 'RUN_LOGS_QUEUE_KE
 
 defined('MAPS_UID_FID') or define("MAPS_UID_FID", 'MAPS_UID_FID');                         //登陆用户UID-FID MAP  HAS表
 defined('MAPS_FID_UID') or define("MAPS_FID_UID", 'MAPS_FID_UID');                         //登陆用户FID-UID MAP  HAS表
-defined('MAPS_USER_TOKEN') or define("MAPS_USER_TOKEN", 'MAPS_USER_TOKEN');                //登陆用户TOKEN-UID MAP  HAS表
+defined('MAPS_TOKEN_UID') or define("MAPS_TOKEN_UID", 'MAPS_TOKEN_UID');                   //登陆用户TOKEN-UID MAP  HAS表
+defined('MAPS_UID_TOKEN') or define("MAPS_UID_TOKEN", 'MAPS_UID_TOKEN');                   //登陆用户UID-TOKEN MAP  HAS表
+
 defined('ON_OPEN') or define("ON_OPEN", 'ON_OPEN');                                         //接入时 HAS表
 
 defined('TIME_RSYNC_KEY') or define("TIME_RSYNC_KEY", 'TIME_RSYNC_KEY');                     //时间同步的写获取者
 
 defined('MSG_REDIS_SUBSCRIBE') or define("MSG_REDIS_SUBSCRIBE", 'MSG_REDIS_SUBSCRIBE');      //redis 消息订阅key
-
+defined('MSG_INDEX') or define("MSG_INDEX", 'MSG_INDEX_QIPAI');                                     //redis 消息订阅key首页中奖消息
 ////
 
 

websockServ/app/logic/CmdProxy.php

@@ -16,6 +16,8 @@ use app\lib\DataPack;
 use  app\logic\cmdpro\CmdTest as CmdTest;
 use  app\logic\cmdpro\CmdQuery as CmdQuery;
 use  app\logic\cmdpro\CmdLogin as CmdLogin;
+use  app\logic\cmdpro\CmdSubser as CmdSubser;
+
 
 class CmdProxy
 {
@@ -65,6 +67,9 @@ class CmdProxy
             case "CmdLogin":
                 $obj = new CmdLogin();
                 break;
+            case "CmdSubser":
+                $obj = new CmdSubser();
+                break;
             default:
                 break;
         }
@@ -78,6 +83,7 @@ class CmdProxy
             'login' => 'CmdLogin',
             'query' => 'CmdQuery',
             'test' => 'CmdTest',
+            'subserv' => 'CmdSubser',
         ];
         return $arrs;
     }

websockServ/app/logic/MyServerV2.php

@@ -9,6 +9,7 @@ namespace app\logic;
  * Time: 11:22
  */
 
+use app\lib\DataPack;
 use app\lib\GlobConfigs;
 use Illuminate\Database\Capsule\Manager as Capsule;
 use Illuminate\Database\Capsule\Manager as DB;
@@ -22,7 +23,7 @@ class MyServerV2
     //线程连接实例
     private $workRedis = null;
     private $workMysql = null;
-    private $workPgsql = null ;
+    private $workPgsql = null;
 
     public function __construct()
     {
@@ -78,17 +79,19 @@ class MyServerV2
             MyPgsql::getInstance();
 
             //消息订阅
-            if ($worker_id == 0){
+            if ($worker_id == 0) {
 
-                if ( !$worker_id ){      return ;   }
+                if (!$worker_id) {
+                    return;
+                }
 
-                go(function() use ($serv,$worker_id){
+                go(function () use ($serv, $worker_id) {
                     $config = GlobConfigs::getKey('redis');
                     $redis = new \Swoole\Coroutine\Redis();
                     $redis->setOptions(['compatibility_mode' => true]);
                     $redis->connect($config['host'], $config['port']);
-                    $r = $redis->SUBSCRIBE([MSG_REDIS_SUBSCRIBE,'TEST']);
-                    if ($r){
+                    $r = $redis->SUBSCRIBE([MSG_REDIS_SUBSCRIBE, 'TEST']);
+                    if ($r) {
                         while ($msg = $redis->recv()) {
                             print_r($msg);
                             // msg是一个数组, 包含以下信息
@@ -99,12 +102,10 @@ class MyServerV2
                             if ($type == 'subscribe') // 或psubscribe
                             {
                                 // 频道订阅成功消息，订阅几个频道就有几条
-                            }
-                            else if ($type == 'unsubscribe' && $info == 0) // 或punsubscribe
+                            } else if ($type == 'unsubscribe' && $info == 0) // 或punsubscribe
                             {
                                 break; // 收到取消订阅消息，并且剩余订阅的频道数为0，不再接收，结束循环
-                            }
-                            else if ($type == 'message') // 若为psubscribe，此处为pmessage
+                            } else if ($type == 'message') // 若为psubscribe，此处为pmessage
                             {
                                 /*
                                 if ($need_unsubscribe) // 某个情况下需要退订
@@ -136,6 +137,10 @@ class MyServerV2
 
     public function onMessage($serv, $frame)
     {
+        if ( strtolower($frame->data->data)  == '{"type":"ping"}'){
+                $serv->send($frame->fd,'{"type":"pong"}');
+                return ;
+        }
         Wlog::getInstance()->WriteLog($frame);
         $serv->task($frame);
     }
@@ -144,28 +149,28 @@ class MyServerV2
     {
 
         $token = isset($request->get['token']) ? $request->get['token'] : '';
-        $uid = isset($request->get['uid']) ? $request->get['uid'] : '';
+        //$uid = isset($request->get['uid']) ? $request->get['uid'] : '';
         $fd = $request->fd;
         $redis = $this->workRedis;
         if (!$redis) {
             return;
         }
+        $token_uid = $uid = intval($this->workRedis->hget(MAPS_TOKEN_UID, md5($token)));
+        if (empty($token) || !$token_uid ) {
+            $serv->push($request->fd, DataPack::toJson(['mtype' => 'system_msg', 'stype' => 'invalid_token', 'data' => ['msg' => '无效的token']]));
+            $serv->disconnect($request->fd);
+            return;
+        }
 
-        $redis->hset(ON_OPEN, $fd, json_encode(['fd' => $fd, 'token' => $token, 'uid' => $uid], JSON_UNESCAPED_UNICODE));
-        $str = '你好 !  fd=' . $fd . ' ,ip=' . $request->server['remote_addr'] . ' ,port=' . $request->server['remote_port'] . ' ,time=' . date('Y-m-d H:i:s');
-
-        echo $str ."\n" ;
-
-        $serv->push($fd, $str);
+        $oldfid = $this->workRedis->hget(MAPS_UID_FID, $uid);
+        if ($oldfid != '' && $oldfid != $fd && $serv->exist($oldfid)) {
+            $serv->push($oldfid, DataPack::toJson(['mtype' => 'system_msg', 'stype' => 'force_logout', 'data' => [ 'msg' => '你已在其它地方登陆,本次退出!']]));
+            $serv->disconnect($oldfid);
+        }
 
-        /*
-        $serv->after(120000,function() use($serv,$fd,$redis){
-            if ($serv->exist($fd)){
-                $serv->disconnect($fd,1000,json_encode('未验证用户关闭',JSON_UNESCAPED_UNICODE));
-                $redis->hdel('onOpen',$fd);
-            }
-        });
-        */
+        $this->workRedis->hset(MAPS_UID_FID, $uid, $fd);
+        $this->workRedis->hset(MAPS_FID_UID, $fd, $uid);
+        $serv->push($fd, DataPack::toJson(['mtype' => 'system_msg', 'stype' => 'well_come', 'data' => [ 'msg' => '成功接入']]));
     }
 
     public function onTask($serv, \Swoole\Server\Task $task)
@@ -174,6 +179,7 @@ class MyServerV2
             Wlog::getInstance()->WriteLog($task, 1, $serv->worker_id);
             CmdProxy::getInstance()->ParaCMD($serv, $task);
         } catch (\Exception $e) {
+            Wlog::getInstance()->WriteLog(['onTask error:',$task,$e->getCode() . ' ' . $e->getMessage()], 3, $serv->worker_id);
             echo "发生异常." . $e->getCode() . ' ' . $e->getMessage() . "\n";
         }
     }
@@ -184,8 +190,10 @@ class MyServerV2
 
     public function onClose($serv, $fd, $from_id)
     {
-        $this->workRedis->hdel(ON_OPEN, $fd);
-        echo "Client {$fd} close connection!\n";
+        $uid = $this->workRedis->hget("MAPS_FID_UID", $fd);
+        $this->workRedis->hdel(MAPS_UID_FID, $uid);
+        $this->workRedis->hdel(MAPS_FID_UID, $fd);
+        echo "ClientFd:{$fd} -- uid:{$uid} close connection!\n";
     }
 
 
@@ -269,7 +277,6 @@ class MyServerV2
     }
 
 
-
     //Mysql Ping检测
     private function PingPgsql($serv, $worker_id)
     {
@@ -311,14 +318,17 @@ class MyServerV2
     private function startInit()
     {
         $conf = GlobConfigs::getKey('redis');
+        $adminconf = GlobConfigs::getKey('admin_conf');
         $redis = new \Redis();
         $ret = $redis->connect($conf['host'], $conf['port'], $conf['overtime']);
         if ($ret) {
             $redis->pipeline();
             $redis->del(MAPS_UID_FID);
             $redis->del(MAPS_FID_UID);
-            $redis->del(MAPS_USER_TOKEN);
-            $redis->del(ON_OPEN);
+            //$redis->del(MAPS_TOKEN_UID);
+            //$redis->del(MAPS_UID_TOKEN);
+            $redis->hset(MAPS_TOKEN_UID,$adminconf['admin_uid'],$adminconf['md5']);
+            $redis->hset(MAPS_UID_TOKEN,$adminconf['md5'],$adminconf['admin_uid']);
             $redis->exec();
             $redis->close();
         }

websockServ/app/logic/SubServer.php

@@ -0,0 +1,199 @@
+<?php
+
+namespace app\logic;
+
+/**
+ * Created by PhpStorm.
+ * User: Administrator
+ * Date: 2019/5/20
+ * Time: 11:22
+ */
+
+
+use app\lib\GlobConfigs;
+use  app\lib\Wlog;
+
+use WebSocket\Client as Client;
+
+class SubServer
+{
+    public $serv;
+
+    //线程连接实例
+    private $workRedis = null;
+    private $configs = null;
+
+    public function __construct()
+    {
+        $this->configs = GlobConfigs::getKey('swoolev_subserv');
+        $this->serv = new \swoole_websocket_server($this->configs['host'], $this->configs['port']);
+
+        unset($this->configs['host']);
+        unset($this->configs['port']);
+        $this->serv->set($this->configs);
+
+        $this->serv->on('Start', array($this, 'onStart'));
+        $this->serv->on('WorkerStart', array($this, 'onWorkerStart'));
+        $this->serv->on('message', array($this, 'onMessage'));
+        $this->serv->on('close', array($this, 'onClose'));
+        $this->serv->on('open', array($this, 'onOpen'));
+        $this->serv->on('task', array($this, 'onTask'));
+        $this->serv->on('Finish', array($this, 'onFinish'));
+
+        return $this->serv->start();
+    }
+
+    public function onStart($serv)
+    {
+        echo "Sub on Start " . "\n";
+    }
+
+    public function onWorkerStart($serv, $worker_id)
+    {
+        if ($serv->taskworker) {
+            return;
+        }
+
+        $this->PingRedis($serv, $worker_id);
+        $serv->tick(8000, function () use ($serv, $worker_id) {
+            $this->PingRedis($serv, $worker_id);
+        });
+
+        if ($worker_id == 0) {
+            $serv->tick(1000, function () use ($serv, $worker_id) {
+                $msg_index_aray = $this->getIndexMsg();
+                if (!empty($msg_index_aray)) {
+                    $this->sendIndexMsg($msg_index_aray);
+                }
+            });
+        }
+
+
+    }
+
+    function onWorkerStop(swoole_server $server, int $worker_id)
+    {
+        echo "onWorkerStop.. " . $worker_id . ' ' . "\n";
+        if ($this->workRedis) {
+            $this->workRedis->close();
+        }
+
+    }
+
+
+    public function onMessage($serv, $frame)
+    {
+
+    }
+
+    public function onOpen($serv, $request)
+    {
+
+    }
+
+    public function onTask($serv, \Swoole\Server\Task $task)
+    {
+
+    }
+
+    public function onFinish($serv, $task_id, $data)
+    {
+    }
+
+    public function onClose($serv, $fd, $from_id)
+    {
+
+    }
+
+
+    //redis Ping检测
+    private function PingRedis($serv, $worker_id)
+    {
+        if (empty($this->workRedis)) {
+            $this->ConectToRedis($serv, $worker_id);
+            return true;
+        }
+        $redis = $this->workRedis;
+        if (!$redis) {
+            $ping_ret = false;
+        } else {
+            $ping_ret_s = $redis->ping();
+            $ping_ret = "+pong" === strtolower($ping_ret_s) ? true : false;
+        }
+
+        if (!$ping_ret) {
+            return $this->ConectToRedis($serv, $worker_id);
+        }
+        return true;
+    }
+
+    //工作线程 同步阻塞 redis客户端
+    private function ConectToRedis($serv, $worker_id)
+    {
+        $conf = GlobConfigs::getKey('redis');
+        $redis = new \Redis();
+        try {
+            $ret = $redis->connect($conf['host'], $conf['port'], $conf['overtime']);
+        } catch (\Exception $e) {
+            $ret = false;
+        }
+
+        if ($ret) {
+            Wlog::getInstance()->WriteLog("success:成功建立redis连接[In Subserver] " . date("Y-m-d H:i:s") . ' ' . $worker_id);
+            $this->workRedis = $redis;
+            return $redis;
+        } else {
+            Wlog::getInstance()->WriteLog("error:建立redis连接[In Subserver] " . date("Y-m-d H:i:s") . ' ' . $worker_id);
+            $this->workRedis = null;
+        }
+        return false;
+    }
+
+
+    ////////////////////////
+    //去redis里找 msg_index 是否有数据，有取出来，
+    private function getIndexMsg()
+    {
+        $redis = $this->workRedis;
+        $len = $redis->LLEN(MSG_INDEX);
+
+        if ($len <=0 ) {
+            return;
+        }
+        $datas = [];
+        while ($now = $redis->rpop(MSG_INDEX)) {
+            if (empty($now)) {
+                break;
+            }
+            $datas[] = $now;
+        }
+
+        return $datas;
+    }
+
+    ///如果有msg_index数据，就通知服务端下发
+    private function sendIndexMsg($msg_index_aray)
+    {
+        $conf = GlobConfigs::getKey('admin_conf');
+        $token = $conf['admin_token'];
+        $url = $conf['url'];
+        $client = new Client("$url?token=" . $token);
+
+        $arr = [
+            'cmd' => 'subserv',
+            'act' => 'index_msg',
+            'data' => $msg_index_aray,
+            'time' => time(),
+            'token' => '',
+        ];
+        $str = json_encode($arr, 256);
+        $r = $client->send($str);
+        $reciv = $client->receive();
+        $client->close();
+
+        unset($client);
+    }
+    /////////////////////////
+
+
+}

websockServ/app/logic/cmdpro/CmdSubser.php

@@ -0,0 +1,33 @@
+<?php
+/**
+ * Created by PhpStorm.
+ * User: Administrator
+ * Date: 2019/5/21
+ * Time: 16:57
+ */
+namespace app\logic\cmdpro;
+
+use app\lib\DataPack;
+use app\logic\cmdpro\CmdInf as CmdInf;
+
+class CmdSubser  extends CmdInf
+{
+    public function index_msg($serv, $task)
+    {
+        go(function()use($serv,$task){
+            $pack = json_decode($task->data->data, true);
+            $msgarr = $pack['data'];
+            if (!is_array($msgarr)){
+                return ;
+            }
+            foreach ($msgarr as $msg){
+                $msg = DataPack::toJson(['mtype'=>'system_msg','stype'=>'index_msg','data'=>$msg]);
+                $this->onlineBroad($serv, $msg );
+            }
+        });
+    }
+
+
+
+
+}

websockServ/app/logic/cmdpro/CmdTest.php

@@ -112,7 +112,7 @@ class CmdTest extends CmdInf
     //时间同步用
     public function timeRsync($serv,$task){
         $dataArr = json_decode($task->data->data,true);
-        $msg = DataPack::toJson(['mtype'=>'system','stype'=>'TimeSync','data'=>$dataArr['data']]);
+        $msg = DataPack::toJson(['mtype'=>'system_msg','stype'=>'TimeSync','data'=>$dataArr['data']]);
         $this->onlineBroad($serv,$msg);
         return ;
     }

websockServ/commands/bootSubserv.php

@@ -0,0 +1,15 @@
+<?php
+/**
+ * Created by PhpStorm.
+ * User: Administrator
+ * Date: 2019/5/20
+ * Time: 9:36
+ */
+
+define('START_TIME', microtime(true));
+require __DIR__.'/../vendor/autoload.php';
+\app\lib\boot::init();
+
+$ser = new app\logic\SubServer();
+//$ser->start();
+

websockServ/commands/mysql_poole_client_test.php

@@ -20,7 +20,9 @@ $client->set(array(
 ));
 $rts= $client->connect('192.168.2.200', $config['poole_port'], 10) or die("连接失败");//链接mysql客户端
 
-
+if (!$rts){
+    die("连接发生错误，操作中止\n");
+}
 
 $sql =("select  * from members where id=57");
 $client->send($sql);
@@ -35,6 +37,7 @@ print_r(json_decode($resdata,true)) ;
 
 $sql ="select  * from \"districts\" where id=1";
 $client->send($sql);
+$client->waitall();
 $resdata = $client->recv();
 print_r(json_decode($resdata,true)) ;
 

websockServ/configs/configs.php

@@ -37,20 +37,25 @@ $tmp_config = [
             'db' => 0,
             'overtime' => 1,
         ],
+        //主进程
         'swoole' => [
-            'worker_num' => 4,
+            'host' => '0.0.0.0',
+            'port' => '9090',
+            'worker_num' => 2,
             'daemonize' => true,
             'max_request' => 50,
             'task_enable_coroutine' => true,
             'dispatch_mode' => 2,
             'debug_mode' => 1,
-            'task_worker_num' => 20,
+            'task_worker_num' => 5,
             'log_file' => '../logs/err.log',
         ],
-
+        //数据库代理连接池
         'swoolev2' => [
+            'host' => '0.0.0.0',
+            'port' => '9091',
             'worker_num' => 2,
-            'daemonize' => false,
+            'daemonize' => true,
             'max_request' => 5000,
             'task_enable_coroutine' => true,
             'dispatch_mode' => 2,
@@ -58,6 +63,20 @@ $tmp_config = [
             'task_worker_num' => 10,
             'log_file' => '../logs/errv2.log',
         ],
+        //子进程业务
+        'swoolev_subserv' => [
+            'host' => '0.0.0.0',
+            'port' => '9092',
+            'worker_num' => 2,
+            'daemonize' => true,
+            'max_request' => 5000,
+            'task_enable_coroutine' => true,
+            'dispatch_mode' => 2,
+            'debug_mode' => 1,
+            'task_worker_num' => 1,
+            'log_file' => '../logs/errv3.log',
+        ],
+
         'pgsqlpoole'=>[
             'poole_host' => '0.0.0.0' ,
             'poole_port' => 9091,
@@ -90,10 +109,13 @@ $tmp_config = [
             'db_name' => 'ds_cms',
             'db_user' => 'vali',
             'db_pwd' => '1234',
-
         ],
-
-
+        'admin_conf' =>[
+            'url' => 'ws://192.168.2.200:9090',
+            'admin_uid' => 999999999,
+            'md5'=>'70d8d42d4307e63702b25ddef0129f72',
+            'admin_token' => 'aaminaeyJhb34G6i6JIU6I1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJib3JuIiwiaWF0IjoxNTU4NTk1MjgxLCJleHAiOjE1NTg2MDI0ODEsIm5iZiI6MTU1ODU5NTM0MSwic3ViIjo3NiwianRpIjoiNWIwYjlkNjMtYmQ5My1kNTBkLTA1NzgtZGU0NDEyYjFmZjBlIn0.jaYz1IcJJxNCaEF-ELCcE6JjbFSjKPl0p-whJ76dV1w',
+        ],
     ],
     'dev' => [