最近在学习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">
<!--[if lt IE 9]>
<script src="https://cdn.bootcss.com/html5shiv/3.7.3/html5shiv.min.js"></script>
<script src="https://cdn.bootcss.com/respond.js/1.4.2/respond.min.js"></script>
<![endif]-->
<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;
运行效果
现在访问前端页面，对数据库进行增删改操作测试效果。
1.png
结语
这只是个swoole_websocket练习，熟悉一下，大家可以看看Swoole官方文档进行学习，会有不错的提升，之后会做个即时聊天工具玩玩，Just code for fun ~
