Swoole WebSocket 消息服务系统设计：架构、存储与统计
本方案旨在为基于 Swoole 的 PHP WebSocket 服务提供高并发下的消息推送与计数解决方案。核心内容包括 Token 鉴权与防雪崩缓存策略、异步 HTTP 接入与 Redis 双队列缓冲、MySQL 分表异步存储，以及基于计数器与时间戳分离的用户未读消息统计模型。
系统概述
该 WebSocket 服务主要承担两大核心业务：面向客户端（Android/iOS）的用户消息推送服务，以及内部的客服 IM 消息系统。前端通过 Web 服务交互，后端依赖 Swoole 维持长连接。整体架构分为对外 WebSocket 长连接服务、对内 Http 回调服务以及底层数据存储层。
第1步：鉴权与缓存防雪崩策略
目的：确保请求合法性，同时防止因缓存集中过期导致的数据库雪崩效应。
入口：Swoole WebSocket OnOpen 事件或 Http 请求入口。
操作：
- 服务端接收到携带 Token 的请求，调用用户中心接口验证权限并获取
uid。 - 若验证通过，将 Token 与 UID 的关系存入 Redis。关键参数设置：为避免缓存同一时间过期，根据访问峰值（晚间 21-24 点，约 100w 请求/小时）计算随机偏移量。
原文明确指出，用户中心默认缓存时间为 7300s。为了防止缓存雪崩，本地服务的过期时间公式设计为基准时间加上随机值：
$uid = $redis->get($token);
// 基准3650秒 + 随机1~3000秒，确保总和小于7300s
$expireTime = 3650 + rand(1, 3000);
$uid = OAuth::getUserInfo($token);
if (!empty($uid) && intval($uid) > 0) {
// 存储 Token 到 UID 的映射
$redis->setEx($token, $expireTime, $uid);
}
if($uid && $uid > 0){
// 存储 UID 到 Token 的反向映射
$key = 'token_'.$uid;
$redis->setEx($key, $expireTime, $token);
}
第2步：Http 服务接入与消息生产
目的：安全地接收主站业务变更（如评论、私信）产生的消息，解耦业务逻辑与消息推送。
条件：Http 服务仅允许云服务器内网访问，确保安全。
操作顺序：
- 异步埋点触发：主站发生系统消息、评论等操作时，异步发起 HTTP POST 请求至 Swoole 接口。设置超时时间为 2 秒。
- 重试机制：若首次请求失败（返回 false），休眠 100ms 后进行重试。由于是局域网传输，失败率极低，通常意味着服务异常。
示例代码展示了异步发送评论消息的逻辑：
public function swooleComment($uid, $data)
{
$url = $this->swooleUrl . "/api/comment/message";
$commentUid = empty($data['comment_uid']) ? 0 : $data['comment_uid'];
$msg = [
"uid" => $uid,
"msg" => json_encode(['comment_uid' => $commentUid])
];
// 首次尝试，超时2秒
$res = Curl::posturl($url, http_build_query($msg), $this->_headerQArr, 2);
if ($res === false) {
usleep(100000); // 重试间隔100ms
// 二次尝试
$res = Curl::posturl($url, http_build_query($msg), $this->_headerQArr, 2);
}
return $res;
}
第3步：消息缓冲与双队列处理
目的：解决 WebSocket 未连接时的消息丢失问题，并根据用户活跃度分类存储消息。
痛点：Swoole 在 WebSocket 未建立连接时无法实时通信。因此需要将消息先入队，待用户上线后推送。
操作设计：
- 数据结构：使用 Redis List 结构构建快慢双队列。
- 快队列：存放当前最新、高频交互的消息。
- 延迟队列：若用户超过 1 天未登录，消息转入延迟队列；超过 15 天未登录，则消息被释放（不再强推）。
- 执行频率：鉴于 Redis 集群单节点稳定性考虑及 WebSocket 心跳时间（300s），Crontab 定时任务每 4 分钟执行一次消息投递处理，延迟队列每 6 分钟检查一次。
第4步：MySQL 分表与异步写入
目的：应对 300w+ 用户规模下的消息海量存储压力，保证主线程高性能。
操作：
- 分表策略：基于
uid进行取模分表。示例中采用 128 张表（user_push_msg_0~user_push_msg_127）。 - 异步插入：所有数据库写操作不阻塞主进程，通过
TaskManager异步执行。
protected function _getTableName(int $uid): string
{
// 根据 uid 对 128 取模确定分表名
$tableIndex = intval($uid % 128);
return 'user_push_msg_' . $tableIndex;
}
protected function addAsyncMysql(array $pushMsg, int $uid): ?bool
{
$tableName = $this->_getTableName($uid);
if (empty($pushMsg) || empty($tableName) || empty($uid)) return false;
// 异步执行数据库操作
TaskManager::getInstance()->async(function () use ($pushMsg, $tableName) {
DbManager::getInstance()->invoke(function (ClientInterface $client)
use ($pushMsg, $tableName) {
$model = PushMsgModel::invoke($client, $pushMsg);
$model->tableName($tableName)->save();
}, self::MYSQL_CONN_NAME);
});
}
第5步：用户消息数统计方案
目的：精确统计各分组（全体、作者、签约作者等）用户的未读消息数，特别是群组类消息的复杂累加场景。
难点：传统方式难以准确区分“已读”与“未读”，尤其是多用户群发消息。
两步解决法：
- 增量计数器：在 Http 消息接收端，每接收一条针对某用户的消息，计数器加 1。这代表了“产生但未确认”的消息总量。
- 最后读取时间戳表：新建一张专用统计表
message_read_time，以uid为主键。记录用户最后一次查看消息的时间戳。
统计逻辑：当需要展示未读数时，查询该用户所属分组的总消息量，减去在该用户“最后读取时间”之后产生的新消息量。或使用以下结构更新读取行为：
抓要源文编写（此处为更新操作示例） -- 表结构设计 CREATE TABLE `message_read_log` ( `uid` int(10) unsigned NOT NULL DEFAULT '0', `unixtime` int(10) unsigned NOT NULL DEFAULT '0', PRIMARY KEY (`uid`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='用户查看消息最新时间'; -- 更新方式 -- REPLACE INTO ... -- 保证如果存在则更新时间，不存在则插入
最终结果：用户未读数 = 基础计数器数值 - （在 unixtime 之后新增且属于该用户的消息量）。这种分离存储与计数的方法兼顾了实时性与准确性。
以上就是基于 Swoole 的 WebSocket 消息服务系统方案设计详解，更多关于 Swoole WebSocket 消息服务的资料请关注本站其它相关文章！
本站声明
本文内容仅供参考，不代表本站观点。转载或引用请注明出处。如发现内容问题，请通过联系我们页面反馈。
