即时通讯系统核心模块实现
即时通讯系统核心模块实现:从消息传输到存储检索的全链路设计
在当今数字化时代,即时通讯(IM)系统已成为人们日常沟通、工作协作的基础设施。一个高性能、高可靠的 IM 系统需要妥善解决消息的实时传输、持久化存储、快速检索等核心问题。本文将基于一套实际生产环境的代码实现,详细解析 IM 系统中消息传输服务与存储检索服务的设计思路、技术选型与具体实现,带你深入理解 IM 系统的核心工作原理。
一、系统架构 overview:核心模块与技术栈
在展开具体实现前,我们先梳理这套 IM 系统的核心模块与技术选型。从代码来看,该系统采用微服务架构,将核心功能拆分为消息传输服务与消息存储检索服务,通过标准化接口实现模块间通信。
1.1 核心业务流程
IM 系统的核心业务流程可概括为:
- 消息发送:用户发送消息后,由消息传输服务负责验证、封装并转发给目标用户
- 消息存储:传输服务将消息同步到消息队列,由存储服务消费并持久化到数据库
- 消息检索:用户查询历史消息或关键词搜索时,存储服务从数据库 / 搜索引擎中获取数据并返回
1.2 技术栈选型
为满足高并发、低延迟、可扩展的需求,系统选用了以下技术组件:
- RPC 框架:brpc(百度开源的高性能 RPC 框架,支持高并发场景)
- 关系型数据库:MySQL(存储消息元数据、用户会话关系等结构化数据)
- 搜索引擎:Elasticsearch(简称 ES,用于消息全文检索,支持中文分词)
- 消息队列:RabbitMQ(实现消息异步传输,解耦服务间依赖)
- 服务治理:etcd(提供服务注册与发现,支持动态扩缩容)
- 数据序列化:Protobuf(高效的结构化数据序列化协议)
- 日志系统:自定义 logger(记录系统运行状态与错误信息)
1.3 模块交互关系
各模块通过 “服务注册 - 发现” 机制动态感知对方地址,通过 RPC 接口通信:
- 消息传输服务依赖用户服务获取发送者信息,依赖 MySQL 获取会话成员列表
- 消息存储服务依赖文件服务存储图片 / 语音等二进制数据,依赖用户服务获取发送者信息
- 所有服务通过 etcd 注册自身地址,通过服务发现机制找到依赖服务的地址
二、数据存储层设计:ES 与 MySQL 的协同方案
消息数据的存储是 IM 系统的基础,需要同时满足 “可靠存储” 与 “高效检索” 的需求。系统采用 “MySQL+ES” 的混合存储方案:MySQL 存储消息完整元数据,ES 存储消息索引与文本内容用于检索。
2.1 Elasticsearch 封装:索引设计与操作抽象
ES 作为全文搜索引擎,其核心是索引设计与查询语法。代码中通过ESIndex、ESInsert、ESRemove、ESSearch等类对 ES 操作进行封装,简化上层调用。
2.1.1 ES 索引设计:字段类型与分词策略
索引是 ES 中数据的组织形式,类似 MySQL 的表结构。在 IM 系统中,需要为 “用户” 和 “消息” 分别创建索引,其中消息索引需要支持中文分词检索。
// 用户索引创建示例(ESUser类) bool createIndex() { bool ret = zrt::ESIndex(_es_client, "user") .append("user_id", "keyword", "standard", true) // 用户ID:精确匹配,不分词 .append("nickname") // 昵称:文本类型,默认ik分词 .append("phone", "keyword", "standard", true) // 手机号:精确匹配 .append("description", "text", "standard", false) // 描述:文本类型 .append("avatar_id", "keyword", "standard", false) // 头像ID:精确匹配 .create(); // ... 日志与返回处理 } 索引设计的核心考量:
- 字段类型:区分 “keyword”(精确匹配,如 ID、手机号)与 “text”(全文检索,如昵称、消息内容)
- 分词器:中文场景使用
ik_max_word(最大化分词),英文 / 数字使用standard - 动态映射:开启
dynamic: true,允许新增字段自动映射类型(灵活应对业务扩展)
2.1.2 ES 操作封装:CRUD 接口抽象
为简化 ES 操作,代码通过链式调用封装了索引创建、数据插入、删除、查询等操作:
// 数据插入封装(ESInsert类) template<typename T> ESInsert &append(const std::string &key, const T &val){ _item[key] = val; // 用Json::Value暂存数据 return *this; } bool insert(const std::string) { std::string body; Serialize(_item, body); // 序列化为JSON try { auto rsp = _client->index(_name, _type, id, body); // 调用ES客户端API // 状态码检查与异常处理 } } 查询操作通过ESSearch类实现,支持组合条件查询(must/should/must_not):
// 消息搜索示例(ESMessage类) std::vector<zrt::Message> search(const std::string &key, const std::string &ssid) { Json::Value json_user = ESSearch(_es_client, "message") .append_must_term("chat_session_id.keyword", ssid) // 必须匹配会话ID .append_must_match("content", key) // 必须匹配关键词 .search(); // 结果解析与转换 } 这种封装的优势在于:
- 隐藏 ES 底层 API 细节,上层无需关注 JSON 构建与网络请求
- 链式调用简化多条件查询的组合,代码可读性更高
- 统一异常处理与日志记录,减少重复代码
2.2 MySQL 消息存储:结构化数据的可靠持久化
MySQL 主要存储消息的完整元数据(如发送者 ID、会话 ID、时间戳、文件信息等),支持按会话、时间范围的高效查询。代码中通过MessageTable类封装 MySQL 操作(基于 odb ORM 框架)。
// 消息存储示例(MessageServiceImpl类) ret = _mysql_message->insert(msg); // 插入消息到MySQL // 历史消息查询(基于时间范围) auto msg_lists = _mysql_message->range(chat_ssid, stime, etime); // 最近消息查询(按时间倒序取前N条) auto msg_lists = _mysql_message->recent(chat_ssid, msg_count); MySQL 表设计的核心字段(对应Message类):
message_id:主键,消息唯一标识session_id:会话 ID,用于关联会话user_id:发送者 ID,关联用户信息message_type:消息类型(文本 / 图片 / 文件 / 语音)create_time:发送时间,用于排序与范围查询file_id/file_name/file_size:文件相关元数据(非文本消息)
2.3 存储协同策略:MySQL 与 ES 的分工
为什么需要同时使用 MySQL 和 ES?两者的分工如下:
- MySQL:存储完整消息数据,支持按会话、时间的精确查询,保证数据可靠性(ACID 特性)
- ES:存储消息文本内容与索引,支持全文检索、关键词高亮等高级查询,提供毫秒级响应
数据同步流程:
- 消息到达后,先写入 MySQL 保证数据不丢失
- 仅文本消息同步到 ES(非文本消息无需检索)
- 删除消息时,同时删除 MySQL 记录与 ES 文档
三、消息传输服务:从发送到转发的全流程
消息传输服务(TransmiteService)是 IM 系统的 “交通枢纽”,负责接收用户发送的消息、封装消息元数据、获取目标用户列表并转发,同时将消息同步到消息队列供存储服务处理。
3.1 RPC 接口设计:定义消息传输契约
使用 Protobuf 定义 RPC 接口,明确服务端与客户端的交互规范。消息传输服务的核心接口是GetTransmitTarget,用于获取消息转发目标:
// transmit.proto service MsgTransmitService { rpc GetTransmitTarget(NewMessageReq) returns (GetTransmitTargetRsp); } message NewMessageReq { string request_id = 1; // 请求唯一标识 string user_id = 2; // 发送者ID string chat_session_id = 3; // 会话ID MessageContent message = 4; // 消息内容 } message GetTransmitTargetRsp { string request_id = 1; bool success = 2; string errmsg = 3; MessageInfo message = 4; // 封装后的消息 repeated string target_id_list = 5; // 目标用户列表 } 接口设计的核心原则:
- 包含
request_id用于链路追踪与日志关联 - 明确
success与errmsg字段,便于错误处理 - 消息内容与元数据分离,
MessageInfo包含发送者信息、时间戳等元数据
3.2 消息封装流程:从原始内容到完整消息
用户发送的原始消息(如文本 “你好”)需要封装为包含元数据的MessageInfo,流程如下:
// TransmiteServiceImpl::GetTransmitTarget MessageInfo message; message.set_message_id(uuid()); // 生成唯一消息ID message.set_chat_session_id(chat_ssid); // 关联会话 message.set_timestamp(time(nullptr)); // 记录发送时间 message.mutable_sender()->CopyFrom(rsp.user_info()); // 填充发送者信息(从用户服务获取) message.mutable_message()->CopyFrom(content); // 填充消息内容 消息 ID 生成采用 UUID,保证全局唯一;发送者信息通过调用用户服务的GetUserInfo接口获取,包含用户昵称、头像等展示所需数据。
3.3 目标用户获取:从会话成员表查询
IM 系统中,消息需要转发给会话中的所有成员(除发送者外)。目标用户列表从 MySQL 的chat_session_member表获取:
// 获取会话成员列表 auto target_list = _mysql_session_member_table->members(chat_ssid); ChatSessionMemeberTable封装了会话成员的查询逻辑,返回该会话的所有用户 ID。实际应用中,还需过滤掉发送者自身 ID,避免消息回传。
3.4 消息队列集成:异步存储解耦
为避免消息传输过程被存储操作阻塞,系统通过 RabbitMQ 实现异步存储:传输服务将消息发布到队列,存储服务消费队列消息并持久化。
// 发布消息到RabbitMQ bool ret = _mq_client->publish(_exchange_name, message.SerializeAsString(), _routing_key); 消息队列的作用:
- 解耦:传输服务与存储服务无需直接通信,通过队列间接交互
- 削峰:高并发场景下,队列缓冲消息,避免存储服务被瞬时流量压垮
- 可靠投递:通过 RabbitMQ 的持久化机制,保证消息不会因服务宕机丢失
四、消息存储检索服务:从持久化到高效查询
消息存储检索服务(MessageService)负责消息的持久化存储与查询,提供历史消息查询、近期消息查询、关键词搜索等核心功能,是 IM 系统中数据访问的入口。
4.1 多类型消息处理:文本、图片、文件与语音
IM 系统支持多种消息类型,不同类型的消息存储方式不同:
- 文本消息:内容直接存储到 MySQL 与 ES
- 图片 / 语音 / 文件:二进制数据存储到文件服务,MySQL 仅存储文件 ID 等元数据
// 消息存储处理(MessageServiceImpl::onMessage) switch(message.message().message_type()) { case MessageType::STRING: // 文本消息:直接存储内容 content = message.message().string_message().content(); _es_message->appendData(...); // 同步到ES break; case MessageType::IMAGE: // 图片消息:上传文件到文件服务,存储文件ID ret = _PutFile("", msg.image_content(), ..., file_id); break; // 其他类型消息处理类似 } // 统一存储元数据到MySQL zrt::Message msg(...); msg.file_id(file_id); // 关联文件ID _mysql_message->insert(msg); 文件上传通过调用文件服务的PutSingleFile接口实现,返回的file_id用于后续下载文件时查询。
4.2 历史消息查询:时间范围与分页
用户查询历史消息时,需支持按会话 ID、时间范围筛选,并分页返回结果。实现流程如下:
// GetHistoryMsg接口实现 // 1. 从MySQL查询时间范围内的消息元数据 auto msg_lists = _mysql_message->range(chat_ssid, stime, etime); // 2. 批量获取消息中的文件数据(如图片、语音) std::unordered_set<std::string> file_id_lists; for (const auto &msg : msg_lists) { if (!msg.file_id().empty()) file_id_lists.insert(msg.file_id()); } std::unordered_map<std::string, std::string> file_data_lists; _GetFile(rid, file_id_lists, file_data_lists); // 调用文件服务批量下载 // 3. 批量获取发送者用户信息 std::unordered_set<std::string> user_id_lists; for (const auto &msg : msg_lists) { user_id_lists.insert(msg.user_id()); } std::unordered_map<std::string, UserInfo> user_lists; _GetUser(rid, user_id_lists, user_lists); // 调用用户服务批量获取 // 4. 组装响应数据 for (const auto &msg : msg_lists) { auto message_info = response->add_msg_list(); // 填充消息元数据、用户信息、文件数据 } 优化点:
- 批量查询替代循环单查,减少 RPC 调用次数
- 只下载当前页需要的文件数据,避免大量无效 IO
- 用户信息与文件数据缓存,减少重复查询
4.3 关键词搜索:基于 ES 的全文检索
消息搜索是 IM 系统的高频需求,要求支持关键词匹配、会话内搜索等功能。基于 ES 的实现流程如下:
// MsgSearch接口实现 // 1. 调用ES搜索会话内包含关键词的消息 auto msg_lists = _es_message->search(skey, chat_ssid); // 2. 获取发送者用户信息(同历史消息查询) std::unordered_set<std::string> user_id_lists; for (const auto &msg : msg_lists) { user_id_lists.insert(msg.user_id()); } std::unordered_map<std::string, UserInfo> user_lists; _GetUser(rid, user_id_lists, user_lists); // 3. 组装响应 for (const auto &msg : msg_lists) { auto message_info = response->add_msg_list(); // 填充消息与用户信息 } ES 查询条件构建(ESSearch类):
// 会话内关键词搜索条件 ESSearch(_es_client, "message") .append_must_term("chat_session_id.keyword", ssid) // 限定会话 .append_must_match("content", key) // 匹配关键词 中文搜索优化:
- 使用 IK 分词器对消息内容分词(如 “即时通讯” 分为 “即时”、“通讯”)
- 支持同义词扩展(如 “消息” 与 “信息” 视为等同)
- 按匹配度排序,优先返回相关度高的结果
五、服务治理:注册、发现与连接管理
在微服务架构中,服务的动态注册与发现是保证系统弹性的核心。系统基于 etcd 实现服务治理,通过ServiceManager管理服务连接。
5.1 服务注册:向 etcd 登记服务地址
服务启动时,将自身地址(IP: 端口)注册到 etcd,便于其他服务发现:
// TransmiteServerBuilder::make_registry_object _registry_client = std::make_shared<Registry>(reg_host); _registry_client->registry(service_name, access_host); // 注册服务名与地址 注册信息格式:/services/{service_name}/{instance_id} = {access_host},其中instance_id为服务实例唯一标识(如 UUID)。
5.2 服务发现:监听 etcd 节点变化
服务启动时,通过 etcd 监听依赖服务的地址变化,动态更新可用节点列表:
// MessageServerBuilder::make_discovery_object auto put_cb = std::bind(&ServiceManager::onServiceOnline, _mm_channels.get(), ...); auto del_cb = std::bind(&ServiceManager::onServiceOffline, _mm_channels.get(), ...); _service_discoverer = std::make_shared<Discovery>(reg_host, base_service_name, put_cb, del_cb); ServiceManager维护服务节点的在线状态:
- 当 etcd 新增节点(服务上线),调用
onServiceOnline添加节点 - 当 etcd 删除节点(服务下线),调用
onServiceOffline移除节点
5.3 连接管理:负载均衡与故障转移
ServiceManager的choose方法用于从可用节点中选择一个进行 RPC 调用,实现负载均衡:
// 选择用户服务节点 auto channel = _mm_channels->choose(_user_service_name); if (!channel) { LOG_ERROR("没有可用的用户服务节点"); return false; } UserService_Stub stub(channel.get()); // 创建RPC客户端 负载均衡策略:
- 简单轮询:依次选择节点,适合节点性能相近的场景
- 故障转移:检测到节点不可用时,自动切换到其他节点
- 连接池:复用 TCP 连接,减少握手开销
六、代码设计亮点:可复用与可扩展的实现
这套代码在设计上体现了多个优秀的编程实践,值得借鉴学习。
6.1 Builder 模式:复杂对象的分步构建
服务器的初始化涉及多个组件(MySQL、ES、MQ、RPC 等),参数多且依赖复杂。代码通过TransmiteServerBuilder与MessageServerBuilder实现分步构建:
// 消息存储服务构建示例 MessageServerBuilder builder; // 1. 构建ES客户端 builder.make_es_object({"http://es-node1:9200", "http://es-node2:9200"}); // 2. 构建MySQL客户端 builder.make_mysql_object("user", "pass", "mysql-host", "im_db", "utf8", 3306, 10); // 3. 构建服务发现 builder.make_discovery_object("etcd-host:2379", "/services", "file-service", "user-service"); // 4. 构建消息队列 builder.make_mq_object("rabbit", "pass", "mq-host", "msg-exchange", "store-queue", "store-key"); // 5. 构建RPC服务器 builder.make_rpc_server(8002, 60, 16); // 6. 生成服务器实例 auto server = builder.build(); server->start(); Builder 模式的优势:
- 隐藏对象构建的复杂细节,提供清晰的步骤
- 支持灵活配置,可根据环境(开发 / 测试 / 生产)调整参数
- 便于扩展新组件,无需修改核心逻辑
6.2 接口封装:隔离底层依赖
代码对第三方库(如 elasticlient、brpc、RabbitMQ 客户端)进行了封装,上层业务逻辑不直接依赖具体库的 API:
ESIndex/ESInsert封装 ES 操作,更换 ES 客户端时只需修改封装层MQClient封装 RabbitMQ 的连接、发布、消费操作,屏蔽 AMQP 协议细节ServiceManager封装服务发现与连接管理,更换服务治理组件(如从 etcd 到 Consul)时影响范围小
这种封装降低了代码耦合度,提高了可维护性。
6.3 异常处理与日志:问题排查的关键
系统通过统一的异常处理与日志记录,确保问题可追溯:
// 异常处理示例 try { auto rsp = _client->index(_name, _type, index_id, body); if (rsp.status_code < 200 || rsp.status_code >= 300) { LOG_ERROR("创建ES索引失败,状态码: {}", rsp.status_code); return false; } } catch(std::exception &e) { LOG_ERROR("创建ES索引失败: {}", e.what()); return false; } 日志设计原则:
- 包含
request_id,便于追踪单次请求的全链路 - 区分
DEBUG/INFO/ERROR等级别,生产环境可调整输出级别 - 关键操作(如消息存储、服务调用)必须记录日志,包含输入输出参数
七、优化与扩展:从可用到高性能
这套代码实现了 IM 系统的核心功能,但在高并发、大数据量场景下仍有优化空间。
7.1 性能优化方向
- 缓存热点数据:
- 用户信息、会话列表等高频访问数据缓存到 Redis,减少 MySQL 查询
- 近期消息缓存,减少 ES 与 MySQL 访问
- 异步化处理:
- 非核心流程(如消息已读状态更新)异步处理,降低主流程延迟
- 使用线程池处理批量操作(如批量消息推送)
- 索引优化:
- ES 索引按时间分片(如按天),避免单索引过大
- MySQL 添加合适的索引(如
session_id+create_time联合索引)
- 读写分离:
- MySQL 主从分离,写操作走主库,读操作走从库
- ES 配置副本,查询请求分发到副本节点
7.2 功能扩展建议
- 消息已读状态:
- 增加
read字段标记消息已读状态 - 实现 “已读回执” 功能,同步用户阅读状态
- 消息撤回:
- 支持指定时间内(如 5 分钟)撤回消息
- 撤回时同时删除 MySQL 与 ES 中的数据,并通知接收方
- 离线消息:
- 为离线用户缓存消息,上线后批量推送
- 实现消息同步机制,保证多端数据一致
- 搜索增强:
- 支持按发送者、时间范围过滤搜索结果
- 实现关键词高亮、模糊搜索等高级功能
八、总结
本文基于实际代码实现,详细解析了 IM 系统中消息传输与存储检索服务的设计与实现。从技术选型来看,“MySQL+ES” 的混合存储方案平衡了可靠性与检索效率,“RPC + 消息队列” 的通信方式实现了服务解耦与异步处理,“etcd+ServiceManager” 的服务治理保证了系统的弹性与可扩展性。
代码中体现的 Builder 模式、接口封装、异常处理等设计思想,不仅保证了当前功能的实现,也为后续扩展奠定了基础。在实际应用中,还需根据业务规模与性能需求,进一步优化缓存策略、索引设计与异步处理机制,才能构建出高性能、高可靠的 IM 系统。
IM 系统的核心价值在于 “高效沟通”,而支撑这一价值的,正是这些隐藏在代码中的技术细节与架构设计。希望本文能为你理解 IM 系统的工作原理,或设计自己的通讯系统提供有益的参考。