跟着源码学IM(九)：基于Netty实现一套分布式IM系统

Posted on 2021-12-20 18:21 Jack Jiang 阅读(175) 评论(0) 编辑收藏

本文作者小傅哥，原题“使用DDD+Netty，开发一个分布式IM(即时通信)系统”。为了提升阅读体验，有大量修订和改动，感谢原作者。

0、系列文章

《跟着源码学IM(一)：手把手教你用Netty实现心跳机制、断线重连机制》

《跟着源码学IM(二)：自已开发IM很难？手把手教你撸一个Andriod版IM》

《跟着源码学IM(三)：基于Netty，从零开发一个IM服务端》

《跟着源码学IM(四)：拿起键盘就是干，教你徒手开发一套分布式IM系统》

《跟着源码学IM(五)：正确理解IM长连接、心跳及重连机制，并动手实现》

《跟着源码学IM(六)：手把手教你用Go快速搭建高性能、可扩展的IM系统》

《跟着源码学IM(七)：手把手教你用WebSocket打造Web端IM聊天》

《跟着源码学IM(八)：万字长文，手把手教你用Netty打造IM聊天》

《跟着源码学IM(九)：基于Netty实现一套分布式IM系统》（* 本文）

1、本文引言

计算机编程的学习，能不能把知识学到手，讲究的是动手实践。在我编写的文章中，基本都是以实践代码验证结果为核心来讲述文章内容。

从小我就喜欢动手，就以一个即时通信的项目为例，已经基于不同技术方案实现了5、6次，仅仅为了实践技术，截图如下。

正如上图这样：

1）有些是刚学完Socket和Swing的时候，想动手试试这些技术能不能写个QQ出来；
2）也有的是因为实习培训需要完成的项目，不过在有了一些基础后，一周时间就能写完全部功能；
3）虽然这些项目在现在看上去还是丑丑的界面，以及代码逻辑可能也不是那么完善。但放在学习阶段的每一次实现中，都能为自己带来很多技术上的成长。

那么，这次借本文的机会，将IM实践的机会留给你，希望你能用的上。

接下来的内容，我会为你介绍如何开发一个IM的方方面面，包括系统架构、通信协议、单聊群聊、表情发送、UI事件驱动等，以及全套的实践源码让你可以上手学习。

注：源码在本文“4、本文源码”一节的附件处可下载。

学习交流：

- 即时通讯/推送技术开发交流5群：215477170 [推荐]
- 移动端IM开发入门文章：《新手入门一篇就够：从零开发移动端IM》
- 开源IM框架源码：https://github.com/JackJiang2011/MobileIMSDK

（本文已同步发布于：http://www.52im.net/thread-3789-1-1.html）

2、知识准备

* 重要提示：本文不是一篇即时通讯理论文章，文章内容全部由实战代码组织而成，如果你对即时通讯（IM）技术理论了解的太少，建议先详细阅读：《新手入门一篇就够：从零开发移动端IM》。

可能有人不知道 Netty 是什么，这里简单介绍下：

Netty 是一个 Java 开源框架。Netty 提供异步的、事件驱动的网络应用程序框架和工具，用以快速开发高性能、高可靠性的网络服务器和客户端程序。

也就是说，Netty 是一个基于 NIO 的客户、服务器端编程框架，使用Netty 可以确保你快速和简单的开发出一个网络应用，例如实现了某种协议的客户，服务端应用。

Netty 相当简化和流线化了网络应用的编程开发过程，例如，TCP 和 UDP 的 Socket 服务开发。

以下是几篇有关Netty的入门文章，值得一读：

如果你连Java的NIO都不知道是什么，下面的文章建议优先读一下：

Netty源码和API的在线阅读地址：

3、运行效果

在开始学习之前，先给大家演示下本文配套源码的运行效果（源码在本文“4、本文源码”一节的附件处可下载）。

聊天页面：

添加好友：

消息提醒：

4、本文源码

本文完整代码附件下载：

（请从同步发布链接中下载：http://www.52im.net/thread-3789-1-1.html）

源码的目录结构，如下图所示：

这套 IM 代码分为了三组模块：UI、客户端、服务端。

之所以这样拆分，是为了将UI展示与业务逻辑隔离，使用事件和接口进行驱动，让代码层次更加干净整洁易于扩展和维护。

各模块的作用，具体解释如下：

5、系统设计

在这套IM中，服务端采用DDD领域驱动设计模式进行搭建。将 Netty 的功能交给 SpringBoot 进行启停控制，同时在服务端搭建控制台可以非常方便的操作通信系统，进行用户和通信管理。在客户端的建设上采用UI分离的方式进行搭建，以保证业务代码与UI展示分离，做到非常易于扩展的控制。

另外，在功能实现上包括：完美仿照微信桌面版客户端、登录、搜索添加好友、用户通信、群组通信、表情发送等核心功能。如果有对于实际需要使用的功能，可以按照这套系统框架进行扩展。

解释一下：

1）UI开发：使用JavaFx与Maven搭建UI桌面工程，逐步讲解登录框体、聊天框体、对话框、好友栏等各项UI展示及操作事件；
2）架构设计：使用DDD领域驱动设计的四层模型结构与Netty结合使用，架构出合理的分层框架（相应库表功能的设计）；
3）功能实现：包括；登录、添加好友、对话通知、消息发送、断线重连等各项功能。

6、UI开发

6.1 功能划分

聊天窗体，相对于登陆窗体来说，聊天窗体的内容会比较多，同时也会相对复杂一些。

下图是聊天窗体的功能定义草图：

如上图所示：

1）首先是我们整个聊天主窗体的定义，是一块空白面板，并去掉默认的边框按钮 (最小化、退出等)；
2）之后是我们左侧边栏，我们称之为条形 Bar，功能区域的实现；
3）最后添加窗体事件，当点击按钮时变换内容面板中的填充信息。

6.2 聊天界面

对话框选中后的内容区域展现，也就是用户之间信息发送和展现。

从整体上看这是一个联动的过程，点击左侧的对话框用户，右侧就有相应内容的填充。那么右侧被填充对话列表 ListView 需要与每一个对话用户关联，点击聊天用户的时候，是通过反复切换填充的过程。效果如下图所示。

参见上图，我解释一下：

1）点击左侧的每一个对话框体，右侧聊天框填充内容即随之变化（同时还有相应的对话名称也会也变化）；
2）对话框中左侧展示好友发送的信息，右侧展示个人发送的信息（同时消息内容会随着内容的增多而增加高度和宽度）；
3）最下面是文本输入框，在后面的实现里我们文本输入框采用公用的方式进行设计，当然你也可以设计为单独的个人使用。

6.3 好友列表

大家都经常使用 PC 端的微信，可以知道在好友栏里是分了几段内容的，其中包含：新的朋友、公众号、群组和最下面的好友（功能划分如下图）。

参见上图，我解释一下：

1）最上面的搜索框这部分内容不变，和前面的一样。我们目前使用的方式是 fxml 设计，例如这部分是通用功能，可以抽取出来放到代码中，设计成一个组件元素类；
2）经过我们的分析，在使用 JavaFx 组件开发为基础下，这部分是一种嵌套 ListView，也就是最底层的面板是一个 ListView，好友和群组有各是一个 ListView，这样处理后我们会很方便的进行数据填充；
3）另外这样的结构主要有利于在我们程序运行过程中，如果你添加了好友，那么我们需要将好友信息刷新到好友栏中，而在数据填充的时候，为了更加便捷高效，所以我们设计了嵌套的 ListView（如果还不是特别理解，可以从后续的代码中获得答案）。

6.4 事件定义

在桌面版 UI 开发中，为了能使 UI 与业务逻辑隔离，需要在我们把 UI 打包后提供出操作界面的展示效果的接口以及界面操作事件抽象类。

那么可以按照下图理解：

以上这些接口就是我们目前 UI 为外部提供的所有行为接口，这些接口的一个链路描述就是：打开窗口、搜索好友、添加好友、打开对话框、发送消息。

7、通信设计

7.1 系统架构

在前面我说到更适合的架构，才是符合你当下需要最好的架构。

那么怎么设计需要的架构呢？

之所以这样设计，在这个系统里有如下几点前提：

1）系统在服务端要有 web 页面进行管理通信用户以及服务端的控制和监控；
2）数据库的对象类，不要被外部污染，要有隔离性（比如：你的数据库类暴漏给外部做展示类使用了，那么现在需要增加一个字段，而这个字段又不是你数据库存在的属性。那么这个时候就已经把数据库类污染了）。
3）因为目前都是在 Java 语言下实现 Netty 通信，那么服务端与客户端都会需要使用到通信过程中的协议定义和解析。那么我们需要抽离这一层对外提供 Jar 包（利于重用，不然客户端和服务端复制同样的代码维护，就太恶心了）；
4）接口、业务处理、底层服务、通信交互，要有明确的区分和实现，避免造成混乱难以维护。

结合我们上面这四点的前提，你头脑中有什么模型结构体现了？以及相应的技术栈选择上是否有计划了？

接下来我会介绍两种架构设计的模型，一种是你非常熟悉的 MVC，另外一种是你可能听说过的 DDD 领域驱动设计。

7.2 通信协议

从图稿上来看，我们在传输对象的时候需要在传输包中添加一个“帧标识”以此来判断当前的业务对象是哪个对象，也就可以让我们的业务更加清晰，避免使用大量的 if 语句判断。

协议框架：

agreement
└── src
    ├── main
    │   ├── java
    │   │   └── org.itstack.naive.chat
    │   │       ├── codec
    │   │       │    ├── ObjDecoder.java
    │   │       │    └── ObjEncoder.java
    │   │       ├── protocol
    │   │       │    ├── demo
    │   │       │    ├── Command.java
    │   │       │    └── Packet.java
    │   │       └── util
    │   │             └── SerializationUtil.java
    │   ├── resources
    │   │   └── application.yml
    │   └── webapp
    │       └── chat
    │       └── res
    │       └── index.html
    └── test
         └── java
             └── org.itstack.demo.test
                 └── ApiTest.java

协议包：

public abstract class Packet {
    private final static Map<Byte, Class<? extendsPacket>> packetType = new ConcurrentHashMap<>();
    static{
        packetType.put(Command.LoginRequest, LoginRequest.class);
        packetType.put(Command.LoginResponse, LoginResponse.class);
        packetType.put(Command.MsgRequest, MsgRequest.class);
        packetType.put(Command.MsgResponse, MsgResponse.class);
        packetType.put(Command.TalkNoticeRequest, TalkNoticeRequest.class);
        packetType.put(Command.TalkNoticeResponse, TalkNoticeResponse.class);
        packetType.put(Command.SearchFriendRequest, SearchFriendRequest.class);
        packetType.put(Command.SearchFriendResponse, SearchFriendResponse.class);
        packetType.put(Command.AddFriendRequest, AddFriendRequest.class);
        packetType.put(Command.AddFriendResponse, AddFriendResponse.class);
        packetType.put(Command.DelTalkRequest, DelTalkRequest.class);
        packetType.put(Command.MsgGroupRequest, MsgGroupRequest.class);
        packetType.put(Command.MsgGroupResponse, MsgGroupResponse.class);
        packetType.put(Command.ReconnectRequest, ReconnectRequest.class);
    }
    public static Class<? extends Packet> get(Byte command) {
        return packetType.get(command);
    }

    /**
     * 获取协议指令
     *
     * @return 返回指令值
     */
    public abstract Byte getCommand();
}

7.3 添加好友

从上面的流程图中可以看到，这里包含了两部分内容：搜索好友和添加好友。

当添加完成好友后，好友会出现到我们的好友栏中。

并且这里面我们采用的是单方面同意加好友，也就是你添加一个好友的时候，对方也同样有你的好友信息。

如果你的业务中是需要添加好友并同意的，那么可以在发起好友添加的时候，添加一条状态信息，请求加好友。对方同意后，两个用户才能成为好友并进行通信。

添加好友的样例代码：

public class AddFriendHandler extends MyBizHandler<AddFriendRequest> {
    public AddFriendHandler(UserService userService) {
        super(userService);
    }
    @Override
    public void channelRead(Channel channel, AddFriendRequest msg) {
        // 1. 添加好友到数据库中[A->B B->A]
        List<UserFriend> userFriendList = newArrayList<>();
        userFriendList.add(newUserFriend(msg.getUserId(), msg.getFriendId()));
        userFriendList.add(newUserFriend(msg.getFriendId(), msg.getUserId()));
        userService.addUserFriend(userFriendList);
        // 2. 推送好友添加完成 A
        UserInfo userInfo = userService.queryUserInfo(msg.getFriendId());
        channel.writeAndFlush(newAddFriendResponse(userInfo.getUserId(), userInfo.getUserNickName(), userInfo.getUserHead()));
        // 3. 推送好友添加完成 B
        Channel friendChannel = SocketChannelUtil.getChannel(msg.getFriendId());
        if(null== friendChannel) return;
        UserInfo friendInfo = userService.queryUserInfo(msg.getUserId());
        friendChannel.writeAndFlush(newAddFriendResponse(friendInfo.getUserId(), friendInfo.getUserNickName(), friendInfo.getUserHead()));
    }
}

7.4 消息应答

从整体的流程可以看到：在用户发起好友、群组通信的时候，会触发一个事件行为，接下来客户端向服务端发送与好友的对话请求。

服务端收到对话请求后：如果是好友对话，那么需要保存与好友的通信信息到对话框中。同时通知好友，我与你要通信了。你在自己的对话框列表中，把我加进去。

如果是群组通信：是可以不用这样通知的，因为不可能把还没有在线的所有群组用户全部通知（人家还没登录呢），所以这部分只需要在用户上线收到信息后，创建出对话框到列表中即可。可以仔细理解下，同时也可以想想其他实现的方式。

消息应答样例代码：

public class MsgHandler extends MyBizHandler<MsgRequest> {
    public MsgHandler(UserService userService) {
        super(userService);
    }
    @Override
    public void channelRead(Channel channel, MsgRequest msg) {
        logger.info("消息信息处理：{}", JSON.toJSONString(msg));
        // 异步写库
        userService.asyncAppendChatRecord(newChatRecordInfo(msg.getUserId(), msg.getFriendId(), msg.getMsgText(), msg.getMsgType(), msg.getMsgDate()));
        // 添加对话框[如果对方没有你的对话框则添加]
        userService.addTalkBoxInfo(msg.getFriendId(), msg.getUserId(), Constants.TalkType.Friend.getCode());
        // 获取好友通信管道
        Channel friendChannel = SocketChannelUtil.getChannel(msg.getFriendId());
        if(null== friendChannel) {
            logger.info("用户id：{}未登录！", msg.getFriendId());
            return;
        }
        // 发送消息
        friendChannel.writeAndFlush(newMsgResponse(msg.getUserId(), msg.getMsgText(), msg.getMsgType(), msg.getMsgDate()));
    }
}

7.5 断线重连

从上述流程中我们看到：当网络连接断开以后，会像服务端发送重新链接的请求。那么在这个发起链接的过程，和系统的最开始链接有所区别。断线重连是需要将用户的 ID 信息一同发送给服务端，好让服务端可以去更新用户与通信管道 Channel 的绑定关系。

同时还需要更新群组内的重连信息，把用户的重连加入群组映射中。此时就可以恢复用户与好友和群组的通信功能。

消息应答样例代码：

// Channel 状态定时巡检；3 秒后每 5 秒执行一次
scheduledExecutorService.scheduleAtFixedRate(() -> {while(!nettyClient.isActive()) {System.out.println("通信管道巡检：通信管道状态"+ nettyClient.isActive());
        try{System.out.println("通信管道巡检：断线重连 [Begin]");
            Channel freshChannel = executorService.submit(nettyClient).get();
            if(null== CacheUtil.userId) continue;
            freshChannel.writeAndFlush(newReconnectRequest(CacheUtil.userId));
        } catch(InterruptedException | ExecutionException e) {System.out.println("通信管道巡检：断线重连 [Error]");}
    }
}, 3, 5, TimeUnit.SECONDS);

相关文章学习：

7.6 集群通信

如上图所示，我是这样实现IM集群通信的：

1）跨服务之间案例采用redis的发布和订阅进行传递消息，如果你是大型服务可以使用zookeeper；
2）用户A在发送消息给用户B时候，需要传递B的channeId，以用于服务端进行查找channeId所属是否自己的服务内；
3）单台机器也可以启动多个Netty服务，程序内会自动寻找可用端口。

8、本文小结

此IM系统涉及到的技术栈内容较多：Netty4.x、SpringBoot、Mybatis、Mysql、JavaFx、layui等技术栈的使用，以及整个系统框架结构采用DDD四层架构+Socket模块的方式进行搭建，所有的UI都以前后端分离事件驱动方式进行设计。在这个过程中只要你能坚持学习下来，那么一定会收获非常多的内容。足够吹牛啦！

任何一个新技术栈的学习过程都会包括这样一条路线：运行HelloWorld、熟练使用API、项目实践以及最后的深度源码挖掘。那么在听到这样一个需求时候，Java程序员肯定会想到一些列的技术知识点来填充我们项目中的各个模块（例如：界面用JavaFx、Swing等，通信用Socket或者知道Netty框架、服务端控制用MVC模型加上SpringBoot等）。但是怎么将这些各个技术栈合理的架设出我们的系统确是学习、实践、成长过程中最重要的部分。

好了，IM开发实际上涉及的知识维度非常多，限于篇幅就不在这里啰嗦更多，各位读者务必对着源码同步进行学习，这样效果会更好（源码在本文“4、本文源码”一节的附件处可下载）。