如何使用websockets实现跨服务器MySQL数据库的实时数据同步

整体架构思路 #

采用WebSocket长连接推拉结合的架构：由服务器B作为WebSocket服务端监听连接，服务器A作为客户端，在数据产生变更时主动推送变更事件到B端，B端校验通过后直接写入本地数据库，实现毫秒级的实时同步，完全替代原有定时拉取的准实时方案。 #

1. 服务器B端（支持全PHP技术栈）实现 #

直接使用Laravel生态自带的Reverb作为WebSocket服务端，无需额外引入第三方服务：

• 安装配置Laravel Revert后启动服务，监听指定端口，创建专属私有同步频道，必须配置IP白名单+签名校验，仅允许服务器A的IP接入，避免非法推送请求。
• 编写同步事件监听逻辑：收到推送事件后先校验签名合法性，再按照事件类型（新增/更新/删除）执行对应本地库CURD操作。
• 配置失败重试机制：执行写入失败的事件自动存入异步队列，后台脚本自动重试，所有异常操作落盘留痕，避免数据丢失。

2. 服务器A端（仅支持原生PHP）实现 #

完全使用PHP内置函数编写WebSocket客户端，无需任何依赖、框架：

• 在所有涉及MySQL数据变更的业务逻辑后追加同步钩子，无需修改原有业务代码，仅在数据写入完成后触发推送逻辑。
• 原生WebSocket客户端参考实现（无需任何扩展）：

<?php
// 约定的签名密钥，两端提前配置一致
const SYNC_SECRET = '自定义的随机签名密钥';
// 服务器B WebSocket服务配置
const WS_HOST = '服务器B的IP地址';
const WS_PORT = 8080;
const WS_CHANNEL = '/mysql-sync-channel';

// 待同步的变更数据，根据实际业务组装
$changeData = [
    'table' => '操作的表名',
    'op_type' => 'add/edit/delete',
    'primary_key' => '数据主键值',
    'change_fields' => ['变更字段键值对'],
    'event_id' => uniqid().mt_rand(1000,9999), // 唯一事件ID用于去重
    'timestamp' => time()
];
// 生成签名防篡改
$changeData['sign'] = md5(json_encode($changeData).SYNC_SECRET);

// 建立TCP连接
$socket = fsockopen(WS_HOST, WS_PORT, $errno, $errstr, 2);
if (!$socket) {
    // 连接失败写入本地临时文件，后续定时脚本重试推送
    file_put_contents('/tmp/mysql_sync_failed.log', json_encode($changeData).PHP_EOL, FILE_APPEND);
    return false;
}

// 发送WebSocket握手请求
$wsKey = base64_encode(openssl_random_pseudo_bytes(16));
$handshakeHeader = "GET ".WS_CHANNEL." HTTP/1.1\r\n";
$handshakeHeader .= "Host: ".WS_HOST.":".WS_PORT."\r\n";
$handshakeHeader .= "Upgrade: websocket\r\n";
$handshakeHeader .= "Connection: Upgrade\r\n";
$handshakeHeader .= "Sec-WebSocket-Key: ".$wsKey."\r\n";
$handshakeHeader .= "Sec-WebSocket-Version: 13\r\n\r\n";
fwrite($socket, $handshakeHeader);

// 校验握手响应
$response = fread($socket, 1500);
if (!preg_match('#Sec-WebSocket-Accept:\s(.*)$#mU', $response, $matches)) {
    fclose($socket);
    return false;
}

// 封装WebSocket数据帧
$payload = json_encode([
    'event' => 'mysql-sync-event',
    'data' => $changeData,
    'channel' => 'mysql-sync-channel'
]);
$payloadLen = strlen($payload);
$frame = chr(129);
if ($payloadLen <= 125) {
    $frame .= chr(128 | $payloadLen);
} elseif ($payloadLen <= 65535) {
    $frame .= chr(128 | 126) . pack('n', $payloadLen);
} else {
    $frame .= chr(128 | 127) . pack('J', $payloadLen);
}
// 客户端发送数据必须加掩码
$mask = openssl_random_pseudo_bytes(4);
$frame .= $mask;
for ($i = 0; $i < $payloadLen; $i++) {
    $frame .= chr(ord($payload[$i]) ^ ord($mask[$i % 4]));
}

// 发送数据后关闭连接
fwrite($socket, $frame);
fclose($socket);
return true;
?>

• 新增兜底重试脚本：每分钟扫描本地/tmp/mysql_sync_failed.log的失败推送记录，重试推送成功后删除对应条目，避免网络波动导致数据丢失。

3. 数据一致性保障措施 #

• 服务器B端收到事件后先根据唯一event_id做去重判断，避免重复写入导致数据异常。
• 每日凌晨跑低峰校验脚本，对比两端数据库的表行数、主键范围、关键字段校验和，补全不一致的遗漏数据。
• 数据量较大的场景可开启批量推送，攒够10-50条变更后一次性推送，降低连接建立开销。

内容的提问来源于stack exchange，提问作者winfred adrah