Swoole WebSocket 实现mysql实时数据展示

快来打我* 2022-05-15 14:24 354阅读 0赞

最近在学习swoole，自己完成下论坛里留下的作业，通过swoole\_websocket实时展示mysql数据，有个遗留问题，**如何判断mysql是否有增删改**，我想到的方法有：  
1、在应用层中有增删改时，发一个消息到消息队列来监听。  
2、解析mysql bin-log日志。  
3、触发器。  
现在的解决办法是利用触发器，例子中我只监听了一个表，当有多个表时，需要添加大量触发器，触发器本身占用很多资源，太多触发器也不好管理，之后会再想办法解析下bin-log日志来监听Mysql数据变化。

### 服务端代码 ###

<?php
    /**
     * Created by PhpStorm.
     * User: qyc
     * Date: 2017/10/22
     * Time: 下午4:40
     */
    
    define('HOST', '0.0.0.0');
    define('PORT', '9502');
    define('WORKER_NUM', 8);
    define('DISPATCH_MODE', 2);
    define('DAEMONIZE', 0);
    
    class swoole_ws
    {
        private $pdo;
    
        private $server;
    
        //这里写死了数据表，之后完成数据库监听后改写这里逻辑
        private $table
            = [
                'swoole_test',
            ];
    
        public function __construct()
        {
            $this->server = new swoole_websocket_server(HOST, PORT);
            $this->server->set(
                [
                    //开启woker进程数
                    'worker_num'    => WORKER_NUM,
                    //请求分发策略，
                    'dispatch_mode' => DISPATCH_MODE,
                    //是否守护进程
                    'daemonize'     => DAEMONIZE,
                ]
            );
            $this->server->on('message', [$this, 'onMessage']);
            $this->server->on('open', [$this, 'onOpen']);
            $this->server->on('workerStart', [$this, 'onWorkerStart']);
            $this->server->start();
        }
    
        //woker进程
        public function onWorkerStart(swoole_websocket_server $server, $worker_id)
        {
            if ($worker_id == 0) {
                // 0 worker进程开启一个定时器去监听mysql数据变化
                $this->server->tick(500, [$this, 'onTick']);
            }
            //每个worker进程维持一个pdo连接
            $this->pdo = new PDO("mysql:host=localhost;dbname=swoole_ws", "root", "root");
        }
    
        //接受客户端数据
        public function onMessage(swoole_websocket_server $server, swoole_websocket_frame $frame)
        {
    //        echo $frame->data;   //客户端发送的数据
    //        echo $server->worker_id;
        }
    
        //客户端 服务端建立连接并完成握手后的回调
        public function onOpen(swoole_websocket_server $server, swoole_http_request $request)
        {
            //第一次连接去获取一下Mysql数据
            $this->update();
        }
    
        private function update()
        {
            $res = [];
            foreach ($this->table as $table) {
                $res[$table] = $this->getTableData($table);
            }
            foreach ($this->server->connections as $connection) {
                //向所有客户端发送数据
                $this->server->push($connection, json_encode($res));
            }
        }
    
        //获取表数据
        private function getTableData($table)
        {
            $sql = 'select * from ' . $table;
            try {
                $stmt = $this->pdo->prepare($sql);
                $stmt->execute();
                $res = $stmt->fetchAll(PDO::FETCH_ASSOC);
    
                return $res ?: [];
            } catch (Exception $e) {
                return [];
            }
        }
    
        //定时器回调
        public function onTick()
        {
            /*
             * is_update表记录表是否更新过
             * swoole_test表添加了3个触发器， after insert、update、delete，
             * 会向is_update表更新swoole_test是否有更新操作
             */
            try {
                $sql = 'select is_update from is_update where table_name = "swoole_test"';
                $stmt = $this->pdo->prepare($sql);
                $stmt->execute();
                $res = $stmt->fetchAll(PDO::FETCH_ASSOC);
                if ( ! $res) {
                    return [];
                }
                if ($res[0]['is_update'] == 1) {
                    //当is_update字段为1时表明数据有更新，向客户端推送消息
                    $this->update();
                    //更新下表更新字段
                    $update = 'update is_update set is_update=0 where table_name = "swoole_test"';
                    $stmt = $this->pdo->prepare($update);
                    $stmt->execute();
                }
            } catch (Exception $e) {
                $this->pdo = new PDO("mysql:host=localhost;dbname=swoole_ws", "root", "root");
            }
        }
    }
    
    new swoole_ws();

### 客户端代码 ###

<!DOCTYPE html>
    <html lang="zh-CN">
    <head>
        <meta charset="utf-8">
        <meta http-equiv="X-UA-Compatible" content="IE=edge">
        <meta name="viewport" content="width=device-width, initial-scale=1">
        <title>Mysql实时数据</title>
        <link rel="stylesheet" href="https://cdn.bootcss.com/bootstrap/3.3.7/css/bootstrap.min.css"
              integrity="sha384-BVYiiSIFeK1dGmJRAkycuHAHRg32OmUcww7on3RYdg4Va+PmSTsz/K68vbdEjh4u" crossorigin="anonymous">
        
        <script src="https://cdn.bootcss.com/jquery/1.12.4/jquery.min.js"></script>
        <script src="https://cdn.bootcss.com/bootstrap/3.3.7/js/bootstrap.min.js"
                integrity="sha384-Tc5IQib027qvyjSMfHjOMaLkfuWVxZxUPnCJA7l2mCWNIpG9mGCD8wGNIcPD7Txa"
                crossorigin="anonymous"></script>
    </head>
    <body>
    <div class="container">
        <div class="panel panel-default">
            <div class="panel-heading">
                <h3 class="panel-title">swoole_test</h3>
            </div>
            <div class="panel-body">
                <table class="table table-bordered  table-hover">
                    <thead id="thead">
                    </thead>
                    <tbody id="tbody">
                    </tbody>
                </table>
            </div>
        </div>
    </div>
    <script>
        var websocket = new WebSocket('ws://127.0.0.1:9502');
    
        websocket.onopen = function (event) {
            console.log('websocket connect');
            websocket.send('hello swoole');
        };
    
        websocket.onclose = function (event) {
            console.log('websocket close');
        };
    
        websocket.onerror = function (event, e) {
            console.log('error occured:' + event.data);
        };
    
        websocket.onmessage = function (event) {
    
            data = JSON.parse(event.data)['swoole_test'];
            var th = '<tr>';
            var width = 1 / json_length(data[0]) * 100;
            for (var key in data[0]) {
                th += "<th style='width:" + width + "%'>" + key + "</th>";
            }
            th += '</tr>';
            $("#thead").html(th);
    
            var tbody = '';
            for (var line in data) {
                tbody += '<tr>';
                var td = '';
                for (var column in data[line]) {
                    td += "<td>" + data[line][column] + "</td>";
                }
                tbody += td + '</tr>';
            }
            $("#tbody").html(tbody);
        };
    
        function json_length(json) {
            var length = 0;
            for (var item in json) {
                length++;
            }
            return length;
        }
    </script>
    </body>
    </html>

### 触发器 ###

DELIMITER $$
    /*更新触发器*/
    DROP TRIGGER IF EXISTS swoole_test_is_update;
    $$
    CREATE TRIGGER swoole_test_is_update AFTER UPDATE ON swoole_test FOR EACH ROW
    BEGIN
      UPDATE `is_update`
      SET    is_update = 1
      WHERE  TABLE_NAME = 'swoole_test';
    END;
    $$
    /*添加触发器*/
    DROP TRIGGER IF EXISTS swoole_test_is_insert;
    $$
    CREATE TRIGGER swoole_test_is_insert AFTER UPDATE ON swoole_test FOR EACH ROW
    BEGIN
      UPDATE `is_update`
      SET    is_update = 1
      WHERE  TABLE_NAME = 'swoole_test';
    END;
    $$
    /*删除触发器*/
    DROP TRIGGER IF EXISTS swoole_test_is_delete;
    $$
    CREATE TRIGGER swoole_test_is_delete AFTER UPDATE ON swoole_test FOR EACH ROW
    BEGIN
      UPDATE `is_update`
      SET    is_update = 1
      WHERE  TABLE_NAME = 'swoole_test';
    END;
    $$
    DELIMITER ;

### 开启websocker服务端 ###

php swoole_ws;

### 运行效果 ###

现在访问前端页面，对数据库进行增删改操作测试效果。

![Image 1][]

1.png

### 结语 ###

这只是个swoole\_websocket练习，熟悉一下，大家可以看看[Swoole官方文档][Swoole]进行学习，会有不错的提升，之后会做个即时聊天工具玩玩，Just code for fun ~

参考文献：

[https://www.jianshu.com/p/9f27a6d7c29a][https_www.jianshu.com_p_9f27a6d7c29a]

[Image 1]: 
[Swoole]: https://link.jianshu.com?t=https://wiki.swoole.com/
[https_www.jianshu.com_p_9f27a6d7c29a]: https://www.jianshu.com/p/9f27a6d7c29a