ggrtg

app/command/Webchat.php

@@ -0,0 +1,126 @@
+<?php
+declare (strict_types = 1);
+
+namespace app\command;
+
+use think\console\Command;
+use think\console\Input;
+use think\console\input\Argument;
+use think\console\input\Option;
+use think\console\Output;
+use Workerman\worker;
+use Workman\Connection\TcpConnection;
+
+require __DIR__.'/../../vendor/autoload.php';
+
+class Webchat extends Command
+{
+    protected function configure()
+    {
+        // 指令配置
+        $this->setName('startWebchat')
+            ->setDescription('the startWebchat command');
+    }
+
+    protected function execute(Input $input, Output $output)
+    {
+        // 指令输出
+        $output->writeln('startWebchat');
+
+        // 开启一个websocket服务，该端口为前端请求的端口
+        $ws_worker = new Worker('websocket://0.0.0.0:2888');
+        $ws_worker->count = 4;
+        $ws_worker->name='test';
+
+        $ws_worker->onWorkerStart = function() {
+
+            // 开启一个内部端口，方便内部系统推送数据，Text协议格式 文本+换行符
+            // 开启内部端口的目的是方便服务端在某个时刻需要向前端推数据，提供了一个推数据的通道，这里的ip/端口号和前端没有关系，正常情况下是为了实现数据突变上送
+
+            $inner_text_worker = new Worker('text://0.0.0.0:2346');
+
+            // 这里是收集服务端本身向这个ws服务发数据时随时接受，然后处理完成后返回给前端
+            $inner_text_worker->onMessage = function($connection, $buffer) {
+                global $ws_worker;
+                // $data数组格式，里面有uid，表示向那个uid的页面推送数据
+                $data = json_decode($buffer, true);
+                if(!empty($data['uid'])) {
+                    $uid = $data['uid'];
+                    // 通过workerman，向uid的页面推送数据
+                    $ret = $this->sendMessageByUid($ws_worker,$uid, $buffer);
+                } else {
+                    // 广播消息
+                    $ret = $this->broadcast($ws_worker,$buffer);
+                }
+
+                // 返回推送结果给请求这个地址的服务端
+                $connection->send($ret ? 'ok' : 'fail');
+            };
+
+            // ## 执行监听 ##
+
+            $inner_text_worker->listen();
+
+        };
+
+        // 新增加一个属性，用来保存uid到connection的映射
+        $ws_worker->uidConnections = array();
+
+        // 当有客户端发来消息时执行的回调函数
+        $ws_worker->onMessage = function($connection, $data) {
+
+            global $ws_worker;
+
+            // 判断当前客户端是否已经验证,既是否设置了uid
+
+            if(!isset($connection->uid))
+            {
+                // 没验证的话把第一个包当做uid（这里为了方便演示，没做真正的验证）
+                $connection->uid = $data;
+                /* 保存uid到connection的映射，这样可以方便的通过uid查找connection，
+
+                               * 实现针对特定uid推送数据
+
+                               */
+                $ws_worker->uidConnections[$connection->uid] = $connection;
+
+            }
+            // 这句话表示客户端有请求来时，立即给客户端返回数据，正常情况下这里要根据客户端传的参数，做不同处理，返回不同数据
+            $connection->send('receive msg');
+
+        };
+
+        // 当有客户端连接断开时，断开删除已连接的客户端请求
+        $ws_worker->onClose = function($connection) {
+            global $ws_worker;
+
+            if(isset($connection->uid)) {
+                // 连接断开时删除映射
+                unset($ws_worker->uidConnections[$connection->uid]);
+            }
+        };
+
+        worker::runAll();
+    }
+
+    // 针对uid推送数据
+    function sendMessageByUid($worker,$uid, $message) {
+
+        if(isset($worker->uidConnections[$uid])) {
+            $connection = $worker->uidConnections[$uid];
+            $connection->send($message);
+            return true;
+        }
+
+        return false;
+
+    }
+
+    // 向所有验证的用户推送数据
+    function broadcast($worker,$message) {
+        foreach($worker->uidConnections as $connection) {
+            $connection->send($message);
+        }
+        return true;
+    }
+}

app/work/controller/Push.php

@@ -3,6 +3,7 @@ namespace app\work\controller;
 use think\facade\Db;
 use think\worker\Server;
 use Workerman\Lib\Timer;
+use Workerman\worker;
 use think\facade\View;
 
 /*
@@ -14,9 +15,39 @@ class Push extends Server
     protected $socket = 'http://0.0.0.0:2346';   //端口自行修改
 
     protected static $heartbeat_time = 55;
+    public $connectionArr = array();
 
     public function onWorkerStart($worker)
     {
+
+        // 开启一个内部端口，方便内部系统推送数据，Text协议格式 文本+换行符
+        // 开启内部端口的目的是方便服务端在某个时刻需要向前端推数据，提供了一个推数据的通道，这里的ip/端口号和前端没有关系，正常情况下是为了实现数据突变上送
+
+        $inner_text_worker = new Worker('text://0.0.0.0:5678');
+
+        // 这里是收集服务端本身向这个ws服务发数据时随时接受，然后处理完成后返回给前端
+        $inner_text_worker->onMessage = function($connection, $buffer) {
+            //global $worker;
+            // $data数组格式，里面有uid，表示向那个uid的页面推送数据
+            $data = json_decode($buffer, true);
+            if(!empty($data['uid'])) {
+                $uid = $data['uid'];
+                // 通过workerman，向uid的页面推送数据
+                $ret = $this->sendMessageByUid($uid, $buffer);
+            } else {
+                // 广播消息
+                $ret = $this->broadcast($buffer);
+            }
+
+            // 返回推送结果给请求这个地址的服务端
+            $connection->send($ret ? 'ok888' : 'fail666');
+        };
+
+        // ## 执行监听 ##
+        $inner_text_worker->listen();
+
+
+        // 进程启动后设置一个每秒运行一次的定时器
         //查看是否有新的充值或提现订单，有就推送给所有用户
         Timer::add(3, function () use ($worker) {
 
@@ -31,10 +62,27 @@ class Push extends Server
 
 
                 foreach ($worker->connections as $connection) {
+
+                    // 有可能该connection还没收到过消息，则lastMessageTime设置为当前时间
                     if (empty($connection->lastMessageTime)) {
                         $connection->lastMessageTime = $time_now;
+                        continue;
+                    }
+
+                    if((int)$connection->uid !== 10000){
+                        $connection->send("当前用户id非10000");
+                    }else{
+                        $data =  [
+                            'toUser' => 10000,
+                            'msgContent' => 'success6666',
+                            'connection' => json($connection)
+                        ];
+                        //$this->sendMsg($data);
                     }
 
+
+                    //$connection->lastMessageTime 是指上次通讯时间
+                    //// 上次通讯时间间隔大于心跳间隔，则认为客户端已经下线，关闭连接
                     if ($time_now - $connection->lastMessageTime > self::$heartbeat_time) {
                         //后台控制连接
                         //前端断了 可以点击重新连接 类似游戏重新登录
@@ -43,7 +91,7 @@ class Push extends Server
 
                     $data = ['time'=> $time_now,'lastMessageTime' => $connection->lastMessageTime];
 
-                    $connection->send(json_encode($data));
+                    //$connection->send(json_encode($data));
                 }
                 //处理发送状态 ['is_push' => 1]
             } else {
@@ -61,7 +109,113 @@ class Push extends Server
         });
     }
 
+    // 针对uid推送数据
+    function sendMessageByUid($uid, $message)
+    {
+        if (isset($this->connectionArr[$uid])) {
+            $connection = $this->connectionArr[$uid];
+            $connection->send($message);
+            return true;
+        }
+    }
+
+    // 向所有验证的用户推送数据
+    function broadcast($message) {
+        foreach($this->connectionArr as $connection) {
+            $connection->send($message);
+        }
+        return true;
+    }
+
+    /**
+     * 收到信息
+     * @param $connection
+     * @param $data
+     */
+    public function onMessage($connection, $data)
+    {
+        //收到消息刷新最后时间 前端发送心跳
+        $connection->lastMessageTime = time();
+        $wsData = json_decode($data,true);
+
+        if(!isset($wsData['option'])){
+            $connection->send('接收数据:'.$data);
+            return;
+        }
+        switch($wsData['option']){
+            case "login":
+                $connection->uid=$wsData['uid'];
+                //echo $wsData['uid']."上线了";
+                $connection->send($wsData['uid']."上线了");
+                $this->setOnline($connection);
+                break;
+            default:
+                $connection->send('其他数据');
+                break;
+        }
+    }
+
+    //如果是发送消息
+    public function sendMsg($data){
+        /*$data =  [
+            'toUser' => 10000,
+            'msgContent' => '主动发送',
+        ];*/
+        $msgStr = json_encode(array("option"=>"sendMsg","msg"=>"发给".$data['toUser']."的消息:".$data['msgContent']));
+        $this->connectionArr[$data['toUser']]->send($msgStr);
+    }
+
+    //用户上线
+    public function setOnline($connection){
+        $this->connectionArr[$connection->uid] = $connection;
+    }
+
+    /**
+     * 当连接建立时触发的回调函数
+     * @param $connection
+     */
+    public function onConnect($connection)
+    {
+    }
+
+    /**
+     * 当连接断开销毁连接对象节约内存空间
+     * @param $connection
+     */
+    public function onClose($connection)
+    {
+        // 连接断开时删除映射
+        unset($this->connectionArr[$connection->uid]);
+    }
+
+    /**
+     * 当客户端的连接上发生错误时触发
+     * @param $connection
+     * @param $code
+     * @param $msg
+     */
+    public function onerror($connection, $code, $msg)
+    {
+        echo "\r\n error $code $msg\n";
+    }
+
     /*
+     * 阿里云
+     * https://ecs.console.aliyun.com/
+     * kod桌面
+     * http://rong.zx2049.com/
+     * 禅道
+     * http://zentao.dev.zx2049.com/
+     * 宝塔
+     * https://47.120.56.77:25705/
+     * layui手册
+     * http://layui.apixx.net/demo/panel.html
+     * tp5.1手册
+     * https://www.kancloud.cn/manual/thinkphp5_1/354084
+     * tp8.0手册
+     * https://doc.thinkphp.cn/v8_0/shitubianliang.html
+     * 后端
+     * http://console.zx2049.com/admin/index/index.html#
      * 客户端
      * http://workerman.dev.zx2049.com/work/push/index
      */
@@ -76,4 +230,27 @@ class Push extends Server
         return View::fetch();
     }
 
+    /*
+     * 测试
+     * http://workerman.dev.zx2049.com/work/push/test
+     */
+    public function test(){
+
+        $post = request()->post();
+        if(empty($post['info'])){
+            return false;
+        }
+        // 建立socket连接到内部推送端口
+        $client = stream_socket_client('tcp://127.0.0.1:5678', $errno, $errmsg, 1);
+        //dump($client);
+        // 推送的数据，包含uid字段，表示是给这个uid推送
+        $data = array('uid'=>'10000', 'percent'=>'88%','msg'=>$post['info']);
+
+        // 发送数据，注意5678端口是Text协议的端口，Text协议需要在数据末尾加上换行符
+        fwrite($client, json_encode($data,JSON_UNESCAPED_UNICODE)."\n");
+
+        // 读取推送结果
+        echo fread($client, 8192);
+    }
+
 }

app/work/controller/Worker.php

@@ -0,0 +1,14 @@
+<?php
+namespace app\work\controller;
+
+use think\worker\Server;
+
+class Worker extends Server
+{
+    protected $socket = 'http://0.0.0.0:2346';
+
+    public function onMessage($connection,$data)
+    {
+        $connection->send(json_encode($data));
+    }
+}

config/worker_server.php

@@ -19,7 +19,7 @@ return [
     'port'           => 2346, // 监听端口
     'socket'         => '', // 完整监听地址
     'context'        => [], // socket 上下文选项
-    'worker_class'   => 'app\work\controller\Push', // 自定义Workerman服务类名 支持数组定义多个服务
+    'worker_class'   => ['app\work\controller\Push'], // 自定义Workerman服务类名 支持数组定义多个服务
 
     // 支持workerman的所有配置参数
     'name'           => 'thinkphp',

view/work/push/index.html

@@ -14,6 +14,11 @@
     <script src="//unpkg.com/layui@2.9.8/dist/layui.js"></script>
 
 
+    <style>
+        #info_li li{
+            width: 100%;float: left;margin-bottom: 5px;
+        }
+    </style>
 </head>
 
 <body>
@@ -123,23 +128,22 @@
         </div>-->
         <div class="layui-form-item">
             <div class="layui-input-block">
-                <button class="layui-btn" lay-submit lay-filter="formDemo">立即提交</button>
+                <button class="layui-btn" lay-submit lay-filter="formDemo" id="myButton">立即提交</button>
                 <button type="reset" class="layui-btn layui-btn-primary">重置</button>
             </div>
         </div>
     </form>
 </div>
 
-<div class="layui-bg-gray" style="padding: 30px;">
+<div class="layui-bg-gray" style="padding: 30px;width:800px;margin: 0 auto;">
     <div class="layui-row layui-col-space15">
-        <div class="layui-col-md6">
-            <div class="layui-panel">
-                <div style="padding: 50px 30px;">一个面板</div>
-            </div>
-        </div>
-        <div class="layui-col-md6">
-            <div class="layui-panel">
-                <div style="padding: 50px 30px;">一个面板</div>
+        <div class="layui-col-md12">
+            <blockquote class="layui-elem-quote">执行过程:</blockquote>
+            <div class="layui-panel" style="height:300px;overflow-y:scroll;">
+                <div style="padding: 50px 30px;" id="info_li">
+                    <li id="start_li"></li>
+
+                </div>
             </div>
         </div>
     </div>
@@ -249,12 +253,17 @@
         //监听提交
         form.on('submit(formDemo)', function(data){
             //layer.msg(JSON.stringify(data.field));
+            var index = layer.load();
+            $("#myButton").attr("disabled", true);
+
             $.ajax({  // 定义ajax发送请求
                 url:'http://console.zx2049.com/api/ai/uploadCsvFrom',  // 请求发送的地址  有三种填写方式，与form标签的action一致
                 method:'post',  // 请求发送的方式
                 data: data.field,  // 请求携带的数据
                 success:function(res){  // 异步等待，当后端响应成功会回调执行匿名函数，并将数据传递给data参数
                     console.log(res);
+                    layer.close(index);
+                    $("#myButton").attr("disabled", false);
 
                     if(res.code == 0){
                         layer.msg('提交成功');
@@ -277,40 +286,83 @@
 
 
 <script>
-    let ws = new WebSocket("ws://47.120.56.77:2346")
-
-    ws.onopen = function() {
-        //绑定连接事件
-        console.log("连接成功");
-
-        //每30秒发送一次心跳
-        setInterval(function(){
+    layui.use(['element', 'layer'], function(){
+        var $ = layui.jquery
+            ,element = layui.element
+            ,layer = layui.layer;
 
-            // 假设ws是一个WebSocket实例
-            if (ws.readyState !== WebSocket.OPEN) {
-                // 连接不是打开状态，可能是关闭中或已关闭，不应该发送数据
-                console.log('WebSocket is not open.');
+        function isJSON(str) {
+            try {
+                const obj = eval('(' + str + ')');
+                return typeof obj === "object" && obj !== null;
+            } catch (e) {
                 return false;
             }
+        }
+
+        /*
+        var str = '{"uid":"10000","percent":"88%","msg":"相同规则与相同主词下,[建筑职称中级需要什么条件]已存在-不可重复导入!"}';
+        console.log(isJSON((str)));
+        var data = JSON.parse(str);
+        if('msg' in data){
+            var html = '<li>'+data.msg+'</li>';
+            $('#start_li').after(html);
+        }*/
+
+
+        //内部会存在此次连接
+        let ws = new WebSocket("ws://47.120.56.77:2346")
 
-            ws.send(JSON.stringify({'type':"peng"}));
-            console.log('发送心跳...');
+        ws.onopen = function() {
+            //绑定连接事件
+            console.log("连接成功");
 
-        },30000)
 
-    };
+            ws.send(JSON.stringify({'uid': '10000','option':'login'}));
+            console.log('发送登录...');
 
-    ws.onmessage = function(evt) {
-        //绑定收到消息事件
-        data = JSON.parse(evt.data)
-        console.log(data);
-        //这里处理收到的消息, type类型有两种: connectin、deposit如果有deposit要提示有新的订单
+            //每30秒发送一次心跳
+            setInterval(function(){
 
-    };
+                // 假设ws是一个WebSocket实例
+                if (ws.readyState !== WebSocket.OPEN) {
+                    // 连接不是打开状态，可能是关闭中或已关闭，不应该发送数据
+                    console.log('WebSocket is not open.');
+                    return false;
+                }
 
+                ws.send(JSON.stringify({'data': '10000'}));
+                console.log('发送心跳...');
+
+            },30000)
+
+        };
+
+
+
+        ws.onmessage = function(evt) {
+            //绑定收到消息事件
+            if(isJSON(evt.data) === true){
+                var data = JSON.parse(evt.data);
+                if('msg' in data){
+                    var html = '<li>'+data.msg+'</li>';
+                    console.log(html);
+                    $('#start_li').after(html);
+                }
+            }else{
+
+            }
+            console.log(evt.data);
+            //这里处理收到的消息, type类型有两种: connectin、deposit如果有deposit要提示有新的订单
+        };
+
+
+        ws.onclose = function(evt) {
+            //绑定关闭或断开连接事件
+            console.log("连接已关闭");
+        };
+
+
+    });
+</script>
 
-    ws.onclose = function(evt) {
-        //绑定关闭或断开连接事件
-        console.log("连接已关闭");
-    };
-</script>