TeamTalk db_proxy_server CMessageModel详解
公风史
1. 简介
基于mysql和redis，CMessageModel用于管理消息计数和消息详情，计数通过redis存储，详情通过mysql存储。
2. 类与接口
CMessageModel
// 从IMMessage_relationID获取nUserId和nPeerId的前nMsgCnt消息记录，保存到lsMsg，消息详情源自mysql IMMessage_xx
void CMessageModel::getMessage(uint32_t nUserId, uint32_t nPeerId, uint32_t nMsgId,
uint32_t nMsgCnt, list<IM::BaseDefine::MsgInfo>& lsMsg)
// 保存nFromId和nToId之间的消息信息到mysql IMMessage_xx，同时增加redis nFromId和nToId之间未读消息记录数，消息详情源自mysql IMMessage_xx
// 成功返回true 失败返回false
bool CMessageModel::sendMessage(uint32_t nRelateId, uint32_t nFromId, uint32_t nToId, IM::BaseDefine::MsgType nMsgType, uint32_t nCreateTime, uint32_t nMsgId, string& strMsgContent)
// 保存nFromId和nToId之间的语音消息信息，语音信息会先存到mysql IMAudio里，并将语音id作为消息内容将消息信息存到IMMessage_xx
// 成功返回true 失败返回false
bool CMessageModel::sendAudioMessage(uint32_t nRelateId, uint32_t nFromId, uint32_t nToId, IM::BaseDefine::MsgType nMsgType, uint32_t nCreateTime, uint32_t nMsgId, const char* pMsgContent, uint32_t nMsgLen)
// 获取nUserId未读消息信息，未读消息数来自redis unread unread_xx，消息详情来自mysql IMMessage_xx
// nTotalCnt: 保存未读消息数 lsUnreadCount: 保存未读消息信息
void CMessageModel::getUnreadMsgCount(uint32_t nUserId, uint32_t &nTotalCnt, list<IM::BaseDefine::UnreadInfo>& lsUnreadCount)
// 获取nRelateId的消息id，源自redis unread msg_id_xx
uint32_t CMessageModel::getMsgId(uint32_t nRelateId)
// 从mysql IMMessage_xx获取nFromId和nToId最近的消息信息
// nMsgId: 保存消息id strMsgData: 保存消息内容 nMsgType: 保存消息类型
void CMessageModel::getLastMsg(uint32_t nFromId, uint32_t nToId, uint32_t& nMsgId, string& strMsgData, IM::BaseDefine::MsgType& nMsgType, uint32_t nStatus)
// 获取nUserId未读消息总数，存入nTotalCnt，信息源自自redis unread unread_xx
void CMessageModel::getUnReadCntAll(uint32_t nUserId, uint32_t &nTotalCnt)
// 获取nUserId和nPeerId之间消息id在lsMsgId里的消息信息，保存到lsMsg，消息信息源自mysql IMMessage_xx，语音详情源自IMAudio
void CMessageModel::getMsgByMsgId(uint32_t nUserId, uint32_t nPeerId, const list<uint32_t> &lsMsgId, list<IM::BaseDefine::MsgInfo> &lsMsg)
// 重置redis nRelateId的消息id (redis unread msg_id_xx)
// 成功返回true 失败返回false
bool CMessageModel::resetMsgId(uint32_t nRelateId)
MessageContent.cpp：利用CMessageModel用于查询或存储消息详情
// 获取消息详情
void getMessage(CImPdu* pPdu, uint32_t conn_uuid)
// 保存消息详情
void sendMessage(CImPdu* pPdu, uint32_t conn_uuid)
// 获取消息id列表里的消息详情
void getMessageById(CImPdu* pPdu, uint32_t conn_uuid)
// 获取最近消息详情
void getLatestMsgId(CImPdu* pPdu, uint32_t conn_uuid)
MessageCounter.cpp：利用CMessageModel用于查询设置或存储消息计数，另外移动端查询和存储device token也在这里处理
// 获取未读消息信息
void getUnreadMsgCounter(CImPdu* pPdu, uint32_t conn_uuid)
// 重置消息计数
void clearUnreadMsgCounter(CImPdu* pPdu, uint32_t conn_uuid)
// 设置设备token
void setDevicesToken(CImPdu* pPdu, uint32_t conn_uuid)
// 获取设备token
void getDevicesToken(CImPdu* pPdu, uint32_t conn_uuid)
3. redis 和 mysql相关库表
127.0.0.1:6379> select 1
OK
127.0.0.1:6379[1]> HGETALL unread_1
(empty array)
127.0.0.1:6379[1]> HGET 1_im_group_msg count
"3"
127.0.0.1:6379[1]> HGET 1_1_im_user_group count
"3"
127.0.0.1:6379[1]> HGETALL 17_im_group_msg
(empty array)
127.0.0.1:6379[1]> HGET 1_17_im_user_group count
(nil)
127.0.0.1:6379[1]> select 4
OK
127.0.0.1:6379[4]> GET device_1
(nil)
mysql> select * from IMGroupMember;
+----+---------+--------+--------+------------+------------+
| id | groupId | userId | status | created | updated |
+----+---------+--------+--------+------------+------------+
| 1 | 1 | 1 | 0 | 1646811069 | 1646811069 |
| 2 | 1 | 17 | 1 | 1646811069 | 1646811069 |
| 3 | 1 | 18 | 0 | 1646811069 | 1646811069 |
+----+---------+--------+--------+------------+------------+
3 rows in set (0.00 sec)
mysql> select * from IMRelationShip;
+----+---------+-------+--------+------------+------------+
| id | smallId | bigId | status | created | updated |
+----+---------+-------+--------+------------+------------+
| 1 | 1 | 2 | 0 | 1640850654 | 1640850654 |
| 2 | 1 | 17 | 0 | 1645607067 | 1645607067 |
+----+---------+-------+--------+------------+------------+
2 rows in set (0.02 sec)
mysql> select * from IMMessage_2;
+----+----------+--------+------+-------+------------------------------------------------------------------+------+--------+------------+------------+
| id | relateId | fromId | toId | msgId | content | type | status | updated | created |
+----+----------+--------+------+-------+------------------------------------------------------------------+------+--------+------------+------------+
| 1 | 2 | 17 | 1 | 1 | 4BTHpThhZZwF3sdUMnVZeGmYj5niYD/anKYbG8NKfS4= | 1 | 0 | 1645607066 | 1645607066 |
| 2 | 2 | 1 | 17 | 2 | dVZlm5rbfUGMPky0wCa7aEfULgKltc1l8975vqp/5WDgAjGO8xbyD9tUy63YSSIG | 1 | 0 | 1645607234 | 1645607234 |
| 3 | 2 | 17 | 1 | 4 | xOb240V1N2lHz69lHLO25Q== | 1 | 0 | 1645683775 | 1645683775 |
| 4 | 2 | 17 | 1 | 5 | xOb240V1N2lHz69lHLO25Q== | 1 | 0 | 1645683842 | 1645683842 |
| 5 | 2 | 17 | 1 | 1 | LMXtutMwnTHUVhK7mPbXtw== | 1 | 0 | 1646291265 | 1646291265 |
| 6 | 2 | 17 | 1 | 2 | Dh6hiOZn42/8g0nt7YyYKQ== | 1 | 0 | 1646291268 | 1646291268 |
| 7 | 2 | 1 | 17 | 1 | gaErOQ34dT0AApIt2t4eDw== | 1 | 0 | 1646811335 | 1646811335 |
| 8 | 2 | 17 | 1 | 2 | xOb240V1N2lHz69lHLO25Q== | 1 | 0 | 1648468503 | 1648468503 |
+----+----------+--------+------+-------+------------------------------------------------------------------+------+--------+------------+------------+
8 rows in set (0.02 sec)
mysql> select * from IMMessage_2 force index (idx_relateId_status_created) where relateId= 2 and status = 0 and msgId <=2 order by created desc, id desc limit 10;
+----+----------+--------+------+-------+------------------------------------------------------------------+------+--------+------------+------------+
| id | relateId | fromId | toId | msgId | content | type | status | updated | created |
+----+----------+--------+------+-------+------------------------------------------------------------------+------+--------+------------+------------+
| 8 | 2 | 17 | 1 | 2 | xOb240V1N2lHz69lHLO25Q== | 1 | 0 | 1648468503 | 1648468503 |
| 7 | 2 | 1 | 17 | 1 | gaErOQ34dT0AApIt2t4eDw== | 1 | 0 | 1646811335 | 1646811335 |
| 6 | 2 | 17 | 1 | 2 | Dh6hiOZn42/8g0nt7YyYKQ== | 1 | 0 | 1646291268 | 1646291268 |
| 5 | 2 | 17 | 1 | 1 | LMXtutMwnTHUVhK7mPbXtw== | 1 | 0 | 1646291265 | 1646291265 |
| 2 | 2 | 1 | 17 | 2 | dVZlm5rbfUGMPky0wCa7aEfULgKltc1l8975vqp/5WDgAjGO8xbyD9tUy63YSSIG | 1 | 0 | 1645607234 | 1645607234 |
| 1 | 2 | 17 | 1 | 1 | 4BTHpThhZZwF3sdUMnVZeGmYj5niYD/anKYbG8NKfS4= | 1 | 0 | 1645607066 | 1645607066 |
+----+----------+--------+------+-------+------------------------------------------------------------------+------+--------+------------+------------+
6 rows in set (0.00 sec)
4. 测试
获取用户1和用户17的消息详情
//
// test_messagemodel.cpp
// test_messagemodel
//
// Created by blueBling on 22-04-25.
// Copyright (c) 2022年blueBling. All rights reserved.
//
#include "MessageContent.h"
#include "MessageCounter.h"
#include "IM.Message.pb.h"
#include<iostream>
using std::cout;
using std::endl;
int test_getmsg() {
// message IMGetMsgListReq {
// //cmd id: 0x0309
// required uint32 user_id = 1;
// required IM.BaseDefine.SessionType session_type = 2;
// required uint32 session_id = 3;
// required uint32 msg_id_begin = 4;
// required uint32 msg_cnt = 5;
// optional bytes attach_data = 20;
// }
IM::Message::IMGetMsgListReq msg;
msg.set_user_id(1);
msg.set_session_type(IM::BaseDefine::SESSION_TYPE_SINGLE);
msg.set_session_id(17);
msg.set_msg_id_begin(2);
msg.set_msg_cnt(10);
msg.set_attach_data("");
CImPdu pdu;
pdu.SetPBMsg(&msg); // 必须先设置这个，这里会为m_buf分配空间
pdu.SetVersion(1);
pdu.SetFlag(0);
pdu.SetServiceId(IM::BaseDefine::SID_MSG);
pdu.SetCommandId(IM::BaseDefine::CID_MSG_LIST_REQUEST);
pdu.SetSeqNum(65282);
pdu.SetReversed(0);
DB_PROXY::getMessage(&pdu, 1);
return 0;
}
int main(){
test_getmsg();
//这里mysql和redis连接池未释放存在内存泄漏问题，解决方法参考test_dbpool
return 0;
}
测试结果：
17:07:51,269 [INFO ] - <DBPool.cpp>|<339>|<Init>,db pool: teamtalk_master, size: 2
17:07:51,310 [INFO ] - <DBPool.cpp>|<339>|<Init>,db pool: teamtalk_slave, size: 2
17:07:51,511 [INFO ] - <MessageContent.cpp>|<75>|<getMessage>,userId=1, peerId=17, msgId=2, msgCnt=10, count=6
5. 源码
类似资料：