Redis实战之多用户聊天室
2017-09-21 17:28
429 查看
多客户端之间的通信,可采用消息推送和消息拉取两种方法。所谓消息推送,即采用publish/subscribe模式,但该方法要求用户必须在线才能接收到消息;所谓消息拉取模式,即由接收端自己拉取存储在某种邮箱里的消息。
ids:chat
群成员:(zset,包括群内所有成员及其拉取的最新消息ID)
chat:群ID Last_Message_ID 用户ID
消息ID计数器:(string)
ids:msg:群ID
群消息:(zset)
msgs:群ID 消息ID 消息内容
seen:用户ID Last_Message_ID 群ID
Current_Message_ID = get ids:msg:群ID
2.2、修改群成员集合,更新用户从群内拉取的最新消息ID:
zadd chat:群ID Current_Message_ID 用户ID
2.3、修改用户加入的群,更新用户从群内拉取的最新消息ID:
zadd seen:用户ID Current_Message_ID 群ID
zrem chat:群ID 用户ID
3.2、修改用户加入的群:
zrem seen:用户ID 群ID
3.3、清除群消息:
1)获取群成员集合中剩余成员数目:
Number = zcard chat:群ID
2)若 Number <= 0,则删除群消息ID计数器和群内所有消息:
del ids:msg:群ID
zremrangebyrank msgs:群ID 0 -1
3)若 Number > 0,则:
获取群成员集合中最小的score,最小score之前的消息必定已被所有用户拉取:
Min_Score = zrange chat:群ID 0 0 withscores
清除已被所有成员拉取的消息:
zremrangebyscore msgs:群ID 0 Min_Score
New_Message_ID = incr ids:msg:群ID
4.2、将消息添加到群消息集合:
zadd msgs:群ID New_Message_ID 消息内容
<群ID, Last_Message_ID> = zrange seen:用户ID 0 -1 withscores
5.2、遍历所加入的群,拉取所有群消息(并记录最大未读群消息ID:Max_Message_ID):
zrangebyscore msgs:群ID Last_Message_ID+1 inf
5.3、修改群成员集合和用户加入群集合,更新用户从群内拉取的最大消息ID:
zadd chat:群ID Max_Message_ID 用户ID
zadd seen:用户ID Max_Message_ID 群ID
5.4、清除已被所有成员拉取的消息:
获取群成员集合中最小的score,最小score之前的消息必定已被所有用户拉取:
Min_Score = zrange chat:群ID 0 0 withscores
清除已被所有人阅读的消息:
zremrangebyscore msgs:群ID -inf Min_Score
数据结构
1、群
群ID计数器:(string)ids:chat
群成员:(zset,包括群内所有成员及其拉取的最新消息ID)
chat:群ID Last_Message_ID 用户ID
消息ID计数器:(string)
ids:msg:群ID
群消息:(zset)
msgs:群ID 消息ID 消息内容
2、用户
群组:(zset,包括用户加入的群及从群内拉取的最新消息ID)seen:用户ID Last_Message_ID 群ID
原理
1、建群
incr ids:chat2、进群
2.1、获取最新群消息ID:Current_Message_ID = get ids:msg:群ID
2.2、修改群成员集合,更新用户从群内拉取的最新消息ID:
zadd chat:群ID Current_Message_ID 用户ID
2.3、修改用户加入的群,更新用户从群内拉取的最新消息ID:
zadd seen:用户ID Current_Message_ID 群ID
3、退群
3.1、修改群成员集合:zrem chat:群ID 用户ID
3.2、修改用户加入的群:
zrem seen:用户ID 群ID
3.3、清除群消息:
1)获取群成员集合中剩余成员数目:
Number = zcard chat:群ID
2)若 Number <= 0,则删除群消息ID计数器和群内所有消息:
del ids:msg:群ID
zremrangebyrank msgs:群ID 0 -1
3)若 Number > 0,则:
获取群成员集合中最小的score,最小score之前的消息必定已被所有用户拉取:
Min_Score = zrange chat:群ID 0 0 withscores
清除已被所有成员拉取的消息:
zremrangebyscore msgs:群ID 0 Min_Score
4、发送消息
4.1、创建群消息ID:New_Message_ID = incr ids:msg:群ID
4.2、将消息添加到群消息集合:
zadd msgs:群ID New_Message_ID 消息内容
5、拉取消息
5.1、获取用户所加入的群信息(包括群ID和从群内最后拉取的消息ID):<群ID, Last_Message_ID> = zrange seen:用户ID 0 -1 withscores
5.2、遍历所加入的群,拉取所有群消息(并记录最大未读群消息ID:Max_Message_ID):
zrangebyscore msgs:群ID Last_Message_ID+1 inf
5.3、修改群成员集合和用户加入群集合,更新用户从群内拉取的最大消息ID:
zadd chat:群ID Max_Message_ID 用户ID
zadd seen:用户ID Max_Message_ID 群ID
5.4、清除已被所有成员拉取的消息:
获取群成员集合中最小的score,最小score之前的消息必定已被所有用户拉取:
Min_Score = zrange chat:群ID 0 0 withscores
清除已被所有人阅读的消息:
zremrangebyscore msgs:群ID -inf Min_Score
实现
1、建群
int ChatManager::createChat() { // 分配群ID redisReply *pRedisReply = (redisReply *)redisCommand(m_pRedisContext, "INCR ids:chat"); if (pRedisReply == NULL) { return -1; } int nChatID = pRedisReply->integer; freeReplyObject(pRedisReply); return nChatID; }
2、进群
bool ChatManager::joinChat(int nChatID, QString &strUser) { //1、获取最新群消息ID QString strMsgIDKeyName = QString("ids:msg:%1").arg(nChatID); redisReply *pRedisReply = (redisReply *)redisCommand(m_pRedisContext, "GET %s", strMsgIDKeyName.toLocal8Bit().data()); if (pRedisReply == NULL) { return false; } int nLastMsgID = pRedisReply->integer; freeReplyObject(pRedisReply); //2、修改群成员集合,更新用户从群内拉取的最新消息ID QString strChatKeyName = QString("chat:%1").arg(nChatID); pRedisReply = (redisReply *)redisCommand(m_pRedisContext, "ZADD %s %d %s", strChatKeyName.toLocal8Bit().data(), nLastMsgID, strUser.toLocal8Bit().data()); if (pRedisReply == NULL) { return false; } freeReplyObject(pRedisReply); //3、修改用户加入的群,更新用户从群内拉取的最新消息ID QString strSeenKeyName = QString("seen:%1").arg(strUser); pRedisReply = (redisReply *)redisCommand(m_pRedisContext, "ZADD %s %d %d", strSeenKeyName.toLocal8Bit().data(), nLastMsgID, nChatID); if (pRedisReply == NULL) { return false; } freeReplyObject(pRedisReply); return true; }
3、退群
bool ChatManager::leaveChat(int nChatID, QString &strUser) { //1、修改群成员集合 QString strChatKeyName = QString("chat:%1").arg(nChatID); redisReply *pRedisReply = (redisReply *)redisCommand(m_pRedisContext, "ZREM %s %s", strChatKeyName.toLocal8Bit().data(), strUser.toLocal8Bit().data()); if (pRedisReply == NULL) { return false; } freeReplyObject(pRedisReply); //2、修改用户加入的群 QString strSeenKeyName = QString("seen:%1").arg(strUser); pRedisReply = (redisReply *)redisCommand(m_pRedisContext, "ZREM %s %d", strSeenKeyName.toLocal8Bit().data(), nChatID); if (pRedisReply == NULL) { return false; } freeReplyObject(pRedisReply); //清除群消息 //3、获取群成员集合中剩余成员数目 pRedisReply = (redisReply *)redisCommand(m_pRedisContext, "ZCARD %s", strChatKeyName.toLocal8Bit().data()); if (pRedisReply == NULL) { return false; } int nMembers = pRedisReply->integer; freeReplyObject(pRedisReply); if (nMembers <= 0) { //群内无剩余成员 //4、删除群消息ID计数器 QString strMsgIDKeyName = QString("ids:msg:%1").arg(nChatID); pRedisReply = (redisReply *)redisCommand(m_pRedisContext, "DEL %s", strMsgIDKeyName.toLocal8Bit().data()); if (pRedisReply == NULL) { return false; } freeReplyObject(pRedisReply); //5、删除群内所有消息 QString strMsgKeyName = QString("msg:%1").arg(nChatID); pRedisReply = (redisReply *)redisCommand(m_pRedisContext, "ZREMRANGEBYRANK %s 0 -1", strMsgKeyName.toLocal8Bit().data()); if (pRedisReply == NULL) { return false; } freeReplyObject(pRedisReply); } else { //群内还有剩余成员 清除已被所有成员拉取的消息 //6、获取群成员集合中最小的score,最小score之前的消息必定已被所有用户拉取 pRedisReply = (redisReply *)redisCommand(m_pRedisContext, "ZRANGE %s 0 0 WITHSCORES", strChatKeyName.toLocal8Bit().data()); if (pRedisReply == NULL) { return false; } int nScore = atoi(pRedisReply->element[1]->str); freeReplyObject(pRedisReply); //7、清除已被所有成员拉取的消息 QString strMsgKeyName = QString("msg:%1").arg(nChatID); pRedisReply = (redisReply *)redisCommand(m_pRedisContext, "ZREMRANGEBYSCORE %s 0 %d", strMsgKeyName.toLocal8Bit().data(), nScore); if (pRedisReply == NULL) { return false; } freeReplyObject(pRedisReply); } return true; }
4、发送消息
bool ChatManager::sendMsg(int nChatID, QString &strMsgInfo) { //1、创建群消息ID QString strMsgIDKeyName = QString("ids:msg:%1").arg(nChatID); redisReply *pRedisReply = (redisReply *)redisCommand(m_pRedisContext, "INCR %s", strMsgIDKeyName.toLocal8Bit().data()); if (pRedisReply == NULL) { return false; } int nNewMsgID = pRedisReply->integer; freeReplyObject(pRedisReply); //2、将消息添加到群消息集合 QString strMsgKeyName = QString("msg:%1").arg(nChatID); pRedisReply = (redisReply *)redisCommand(m_pRedisContext, "ZADD %s %d %s", strMsgKeyName.toLocal8Bit().data(), nNewMsgID, strMsgInfo.toLocal8Bit().data()); if (pRedisReply == NULL) { return false; } freeReplyObject(pRedisReply); return true; }
5、拉取消息
bool ChatManager::fetchMsg(QString &strUser) { //1、获取用户所加入的群信息(包括群ID和从群内最后拉取的消息ID) QString strSeenKeyName = QString("seen:%1").arg(strUser); redisReply *pSeenRedisReply = (redisReply *)redisCommand(m_pRedisContext, "ZRANGE %s 0 -1 WITHSCORES", strSeenKeyName.toLocal8Bit().data()); if (pSeenRedisReply == NULL) { return false; } //2、遍历所加入的群 for (int nIndex = 0; nIndex < pSeenRedisReply->elements; nIndex = nIndex + 2) { int nChatID = atoi(pSeenRedisReply->element[nIndex]->str); int nNextLastMsgID = atoi(pSeenRedisReply->element[nIndex + 1]->str) + 1; //3、拉取所有群消息 QString strMsgKeyName = QString("msg:%1").arg(nChatID); redisReply *pRedisReply = (redisReply *)redisCommand(m_pRedisContext, "ZRANGEBYSCORE %s %d INF WITHSCORES", strMsgKeyName.toLocal8Bit().data(), nNextLastMsgID); if (pRedisReply == NULL) { freeReplyObject(pSeenRedisReply); return false; } if (0 == pRedisReply->elements) { continue; //没有未读消息 } //4、展示所有未读消息 int nMaxMsgID = 0; for (int nMsgIndex = 0; nMsgIndex < pRedisReply->elements; nMsgIndex = nMsgIndex + 2) { QString strMsgInfo = pRedisReply->element[nMsgIndex]->str; int nMsgID = atoi(pRedisReply->element[nMsgIndex + 1]->str); // 展示消息 if (nMsgID > nMaxMsgID) { nMaxMsgID = nMsgID; } } freeReplyObject(pRedisReply); //5、修改群成员集合 更新用户从群内拉取的最大消息ID QString strChatKeyName = QString("chat:%1").arg(nChatID); pRedisReply = (redisReply *)redisCommand(m_pRedisContext, "ZADD %s %d %s", strChatKeyName.toLocal8Bit().data(), nMaxMsgID, strUser.toLocal8Bit().data()); if (pRedisReply == NULL) { freeReplyObject(pSeenRedisReply); return false; } freeReplyObject(pRedisReply); //6、修改用户加入群集合 更新用户从群内拉取的最大消息ID QString strSeenKeyName = QString("seen:%1").arg(strUser); pRedisReply = (redisReply *)redisCommand(m_pRedisContext, "ZADD %s %d %d", strSeenKeyName.toLocal8Bit().data(), nMaxMsgID, nChatID); if (pRedisReply == NULL) { freeReplyObject(pSeenRedisReply); return false; } freeReplyObject(pRedisReply); //清除已被所有成员拉取的消息 //7、获取群成员集合中最小的score,最小score之前的消息必定已被所有用户拉取 pRedisReply = (redisReply *)redisCommand(m_pRedisContext, "ZRANGE %s 0 0 WITHSCORES", strChatKeyName.toLocal8Bit().data()); if (pRedisReply == NULL) { freeReplyObject(pSeenRedisReply); return false; } if (pRedisReply->elements != 2) { continue; } int nOldMsgID = atoi(pRedisReply->element[1]->str); freeReplyObject(pRedisReply); //8、清除已被所有人阅读的消息 pRedisReply = (redisReply *)redisCommand(m_pRedisContext, "ZREMRANGEBYSCORE %s -INF %d", strMsgKeyName.toLocal8Bit().data(), nOldMsgID); if (pRedisReply == NULL) { freeReplyObject(pSeenRedisReply); return false; } freeReplyObject(pRedisReply); } freeReplyObject(pSeenRedisReply); return true; }
相关文章推荐
- ASP.NET SignalR 与 LayIM2.0 配合轻松实现Web聊天室(九) 之 用 Redis 实现用户在线离线状态消息处理(一)
- Spring Boot实战之netty-socketio实现简单聊天室(给指定用户推送消息)
- STS创建Spring Boot项目实战(Rest接口、数据库、用户认证、分布式Token JWT、Redis操作、日志和统一异常处理)
- mysql颠覆实战课程 - redis存储用户表信息
- ASP.NET SignalR 与 LayIM2.0 配合轻松实现Web聊天室(八) 之 用 Redis 实现用户在线离线状态消息处理
- Spring Boot实战之netty-socketio实现简单聊天室(给指定用户推送消息)
- STS创建Spring Boot项目实战(Rest接口、数据库、用户认证、分布式Token JWT、Redis操作、日志和统一异常处理)
- Spring Boot实战之netty-socketio实现简单聊天室(给指定用户推送消息)
- Redis实战《红丸出品》2.2 strings类型及操作
- ASP.NET Google Maps Javascript API V3 实战基础篇一检测用户位置
- Redis和消息队列使用实战
- [DM实战]家用电器用户行为分析与事件识别
- redis实战 SORT 命令
- redis实战课题
- 实战day06(二)----redis集群
- Redis源码剖析和注释(十五)---- 通知功能实现与实战 (notify)
- Redis集群生产环境高可用方案实战过程
- redis集群实战
- redis进阶6-实战python
- 大数据实战:用户流量分析系统