点击上方“Java知音”，选择“置顶公众号”
技术文章第一时间送达！
作者：yuanrw
juejin.im/post/5d73af31f265da0394022e45
本篇将带大家从零开始搭建一个轻量级的IM服务端，IM的整体设计思路和架构在我的上篇博客中已经讲过了，没看过的同学请点击从零开始开发IM（即时通讯）服务端 。
这篇将给大家带来更多的细节实现。我将从三个方面来阐述如何构建一个完整可靠的IM系统。
可靠性
安全性
存储设计
可靠性
什么是可靠性？对于一个IM系统来说，可靠的定义至少是不丢消息、消息不重复、不乱序，满足这三点，才能说有一个好的聊天体验。
不丢消息
我们先从不丢消息开始讲起。
首先复习一下上一篇设计的服务端架构：
我们先从一个简单例子开始思考：当Alice给Bob发送一条消息时，可能要经过这样一条链路：
client-->connecter
connector-->transfer
transfer-->connector
connector-->client
在这整个链路中的每个环节都有可能出问题，虽然tcp协议是可靠的，但是它只能保证链路层的可靠，无法保证应用层的可靠。
例如在第一步中，connector收到了从client发出的消息，但是转发给transfer失败，那么这条消息Bob就无法收到，而Alice也不会意识到消息发送失败了。
如果Bob状态是离线，那么消息链路就是：
client-->connector
connector-->transfer
transfer-->mq
如果在第三步中，transfer收到了来自connector的消息，但是离线消息入库失败，那么这个消息也是传递失败了。
为了保证应用层的可靠，我们必须要有一个ack机制，使发送方能够确认对方收到了这条消息。
具体的实现，我们模仿tcp协议做一个应用层的ack机制。
tcp的报文是以字节（byte）为单位的，而我们以message单位。
发送方每次发送一个消息，就要等待对方的ack回应，在ack确认消息中应该带有收到的id以便发送方识别。
其次，发送方需要维护一个等待ack的队列。每次发送一个消息之后，就将消息和一个计时器入队。
另外存在一个线程一直轮询队列，如果有超时未收到ack的，就取出消息重发。
超时未收到ack的消息有两种处理方式：
和tcp一样不断发送直到收到ack为止。
设定一个最大重试次数，超过这个次数还没收到ack，就使用失败机制处理，节约资源。例如如果是connector长时间未收到client的ack，那么可以主动断开和客户端的连接，剩下未发送的消息就作为离线消息入库，客户端断连后尝试重连服务器即可。
不重复、不乱序
有的时候因为网络原因可能导致ack收到较慢，发送方就会重复发送，那么接收方必须有一个去重机制。
去重的方式是给每个消息增加一个唯一id。这个唯一id并不一定是全局的，只需要在一个会话中唯一即可。例如某两个人的会话，或者某一个群。如果网络断连了，重新连接后，就是新的会话了，id会重新从0开始。
接收方需要在当前会话中维护收到的最后一个消息的id，叫做lastId。每次收到一个新消息， 就将id与lastId作比较看是否连续，如果不连续，就放入一个暂存队列 queue中稍后处理。
例如：
当前会话的lastId=1，接着服务器收到了消息msg(id=2)，可以判断收到的消息是连续的，就处理消息，将lastId修改为2。
但是如果服务器收到消息msg(id=3)，就说明消息乱序到达了，那么就将这个消息入队，等待lastId变为2后，（即服务器收到消息msg(id=2)并处理完了），再取出这个消息处理。
因此，判断消息是否重复只需要判断msgId>lastId && !queue.contains(msgId)即可。如果收到重复的消息，可以判断是ack未送达，就再发送一次ack。
接收方收到消息后完整的处理流程如下：
伪代码如下：
classProcessMsgNode{/***接收到的消息*/privateMessagemessage;/***处理消息的方法*/privateConsumer<Message>consumer;}publicCompletableFuture<Void>offer(Longid,Messagemessage,Consumer<Message>consumer){if(isRepeat(id)){//消息重复sendAck(id);returnnull;}if(!isConsist(id)){//消息不连续notConsistMsgMap.put(id,newProcessMsgNode(message,consumer));returnnull;}//处理消息returnprocess(id,message,consumer);}privateCompletableFuture<Void>process(Longid,Messagemessage,Consumer<Message>consumer){returnCompletableFuture.runAsync(()->consumer.accept(message)).thenAccept(v->sendAck(id)).thenAccept(v->lastId.set(id)).thenComposeAsync(v->{LongnextId=nextId(id);if(notConsistMsgMap.containsKey(nextId)){//队列中有下个消息ProcessMsgNodenode=notConsistMsgMap.get(nextId);returnprocess(nextId,node.getMessage(),consumer);}else{//队列中没有下个消息CompletableFuture<Void>future=newCompletableFuture<>();plete(null);returnfuture;}}).exceptionally(e->{logger.error("[processreceivedmsg]haserror",e);returnnull;});}
安全性
无论是聊天记录还是离线消息，肯定都会在服务端存储备份，那么消息的安全性，保护客户的隐私也至关重要。
因此所有的消息都必须要加密处理。
在存储模块里，维护用户信息和关系链有两张基础表，分别是im_user用户表和im_relation关系链表。
1.im_user表用于存放用户常规信息，例如用户名密码等，结构比较简单。
2.im_relation表用于记录好友关系，结构如下：
CREATETABLE`im_relation`(`id`bigint(20)COMMENT"关系id",`user_id1`varchar(100)COMMENT"用户1id",`user_id2`varchar(100)COMMENT"用户2id",`encrypt_key`char(33)COMMENT"aes密钥",`gmt_create`timestampDEFAULTCURRENT_TIMESTAMP,`gmt_update`timestampDEFAULTCURRENT_TIMESTAMPONUPDATECURRENT_TIMESTAMP,PRIMARYKEY(`id`),UNIQUEKEY`USERID1_USERID2`(`user_id1`,`user_id2`));
3.user_id1和user_id2是互为好友的用户id，为了避免重复，存储时按照user_id1<user_id2的顺序存，并且加上联合索引。
4.encrypt_key是随机生成的密钥。当客户端登录时，就会从数据库中获取该用户的所有的relation，存在内存中，以便后续加密解密。
5.当客户端给某个好友发送消息时，取出内存中该关系的密钥，加密后发送。同样，当收到一条消息时，取出相应的密钥解密。
客户端完整登录流程如下：
client调用rest接口登录。
client调用rest接口获取该用户所有relation。
client向connector发送greet消息，通知上线。
connector拉取离线消息推送给client。
connector更新用户session。
那为什么connector要先推送离线消息再更新session呢？我们思考一下如果顺序倒过来会发生什么：
用户Alice登录服务器
connector更新session
推送离线消息
此时Bob发送了一条消息给Alice
如果离线消息还在推送的过程中，Bob发送了新消息给Alice，服务器获取到Alice的session，就会立刻推送。这时新消息就有可能夹在一堆离线消息当中推过去了，那这时，Alice收到的消息就乱序了。
而我们必须保证离线消息的顺序在新消息之前。
那么如果先推送离线消息，之后才更新session。在离线消息推送的过程中，Alice的状态就是“未上线”，这时Bob新发送的消息只会入库im_offline，im_offline表中的数据被读完之后才会“上线”开始接受新消息。这也就避免了乱序。
存储设计存储离线消息
当用户不在线时，离线消息必然要存储在服务端，等待用户上线再推送。理解了上一个小节后，离线消息的存储就非常容易了。增加一张离线消息表im_offline，表结构如下：
CREATETABLE`im_offline`(`id`int(11)COMMENT"主键",`msg_id`bigint(20)COMMENT"消息id",`msg_type`int(2)COMMENT"消息类型",`content`varbinary(5000)COMMENT"消息内容",`to_user_id`varchar(100)COMMENT"收件人id",`has_read`tinyint(1)COMMENT"是否阅读",`gmt_create`timestampCOMMENT"创建时间",PRIMARYKEY(`id`));
msg_type用于区分消息类型（chat,ack），content加密后的消息内容以byte数组的形式存储。
用户上线时按照条件to_user_id=用户id拉取记录即可。
防止离线消息重复推送
我们思考一下多端登录的情况，Alice有两台设备同时登陆，在这种并发的情况下，我们就需要某种机制来保证离线消息只被读取一次。
这里利用CAS机制来实现：
1.首先取出所有has_read=false的字段。
2.检查每条消息的has_read值是否为false，如果是，则改为true。这是原子操作。
updateim_offlinesethas_read=truewhereid=${msg_id}andhas_read=false
3.修改成功则推送，失败则不推送。
相信到这里，同学们已经可以自己动手搭建一个完整可用的IM服务端了。更多问题欢迎评论区留言~~
GitHub地址，如果觉得有用，记得给个star：
/yuanrw/IM
推荐阅读(点击即可跳转阅读)
1.SpringBoot内容聚合
2.面试题内容聚合
3.设计模式内容聚合
4.Mybatis内容聚合
5.多线程内容聚合
觉得不错？欢迎转发分享给更多人
我知道你 “在看”
