如何用Netty写一个IM聊天系统

骑猪看日落 2023-01-16 12:29 108阅读 0赞

## 什么是Netty ##

简单概括，Netty是一个Java开源框架，是一个异步的、基于事件驱动的网络应用框架，用以快速开发高性能、高可靠的网络IO程序。Netty 本质是一个 NIO 框架，适用于服务器通讯相关的多种应用场景。

## 准备 ##

能够很好的理解这个project，最好对SpringBoot、Netty、长连接及im即时通讯有相关的提前了解。

## 目录结构及源码 ##

项目源码：[github项目源码][github]，有任何问题也欢迎交流讨论

项目基于gradle，目录如下![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L01lZ3VzdGFzX0pKQw_size_16_color_FFFFFF_t_70][]

*  api:Netty服务端
 *  client:Netty客户端
 *  logic:业务逻辑，提供Netty的基础封装，api与client复用

服务端启动类：NettyImDemoApiApplicationContext
    客户端启动类：NettyImDemoClientApplicationContext

## Netty服务端构建 ##

Netty服务端两个核心类：  
**NettyServer服务端引导类**

public class NettyServer {
    
        /**
         * 读取application.yml配置
         */
        @Value("${netty.port}")
        private Integer port;
        /**
         * Netty Server服务端Channel
         */
        private Channel channel;
    
        private EventLoopGroup bossGroup = new NioEventLoopGroup();
    
        private EventLoopGroup workerGroup = new NioEventLoopGroup();
    
        private final NettyServerHandlerInitializer nettyServerHandlerInitializer;
    
        public NettyServer(NettyServerHandlerInitializer nettyServerHandlerInitializer) {
            this.nettyServerHandlerInitializer = nettyServerHandlerInitializer;
        }
    
        /**
         * 启动Netty Server
         * @throws InterruptedException
         */
        @PostConstruct
        public void start() throws InterruptedException {
            // 创建引导，用于netty启动
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            // 在创建ServerBootstrap类实例前，先创建两个EventLoopGroup，它们实际上是两个独立的Reactor线程池
            // Netty默认线程模型是「主从Reactor多线程模型」，
            serverBootstrap.group(bossGroup, workerGroup)
                    // 指定 Channel为服务端NioServerSocketChannel
                    .channel(NioServerSocketChannel.class)
                    .localAddress(new InetSocketAddress(port))
                    // 设置服务端accept队列大小
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    // TCP Keepalive 机制，实现 TCP 层级的心跳保活功能
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    // 允许较小的数据包的发送，降低延迟
                    .childOption(ChannelOption.TCP_NODELAY, true)
                    // 设置客户端连接上来的 Channel 的处理器为 NettyServerHandlerInitializer
                    // 在每一个客户端与服务端建立完成连接时，服务端会创建一个Channel与之对应, NettyServerHandlerInitializer会进行执行initChannel(Channel c) 方法，进行自定义的初始化
                    .childHandler(nettyServerHandlerInitializer);
            // 绑定端口，并同步等待成功，即启动Netty Server服务端
            ChannelFuture future = serverBootstrap.bind().sync();
            if (future.isSuccess()) {
                // 启动成功，获取Channel引用
                channel = future.channel();
                log.info("start Netty Server port {}", port);
            }
        }
    
        /**
         * 关闭 Netty Server
         */
        @PreDestroy
        public void gracefulShutdown() {
            if (Objects.nonNull(channel)) {
                channel.close();
            }
            // 线程组优雅关闭，例如线程池资源等
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

bossGroup负责服务端接收客户端的连接:  
1.接收客户端TCP连接，初始化Channel参数；  
2.将链路状态变更事件通知给ChannelPipeline；  
为什么主从(两个线程组)优势：创建专门用于接受客户端连接的 bossGroup 线程组，避免因为已连接的客户端的数据读写频繁，影响新的客户端的连接。

workerGroup用于服务端接受客户端的数据读写,负责处理IO相关的读写操作或者执行系统task、定时task等  
1.异步读取远端数据，发送读事件到ChannelPipeline；  
2.异步发送数据到远端，调用ChannelPipeline的发送消息接口；  
3.执行系统调用Task；  
4.执行定时任务Task，如空闲链路检测等；  
为什么多线程优势：创建专门用于接收客户端读写的 workerGroup 线程组，多个线程进行客户端的数据读写，可以支持更多客户端。

这种设计的介绍可以看下「Netty的Reactor线程模型」，默认是主从Reactor多线程模型

**NettyServerHandlerInitializer对新创建的客户端Channel的pipeline初始化配置类**

public class NettyServerHandlerInitializer extends ChannelInitializer<Channel> {
    
        /**
         * 心跳超时时间
         */
        private static final Integer READ_TIMEOUT_SECONDS = 3 * 60;
    
        private final MessageDispatcher messageDispatcher;
        private final NettyServerHandler nettyServerHandler;
    
        public NettyServerHandlerInitializer(MessageDispatcher messageDispatcher, NettyServerHandler nettyServerHandler) {
            this.messageDispatcher = messageDispatcher;
            this.nettyServerHandler = nettyServerHandler;
        }
    
        /**
         * @param ch 此时创建的客户端 Channel
         */
        @Override
        protected void initChannel(Channel ch) {
            ChannelPipeline channelPipeline = ch.pipeline();
            channelPipeline
                    .addLast(new ReadTimeoutHandler(READ_TIMEOUT_SECONDS, TimeUnit.SECONDS))
                    // 自定义协议的编码器
                    .addLast(new InvocationEncoder())
                    // 解码器
                    .addLast(new InvocationDecoder())
                    //  消息处理
                    .addLast(messageDispatcher)
                    // 服务端业务处理
                    .addLast(nettyServerHandler)
                    ;
        }
    }

用于在某个Channel注册到EventLoop后，对这个Channel执行一些初始化操作。ChannelInitializer虽然会在一开始会被注册到Channel相关的pipeline里(继承于ChannelInboundHandler接口, 实际也是一个ChannelHandler)，但是在初始化完成之后，ChannelInitializer会将自己从pipeline中移除，不会影响后续的操作。

## Netty客户端构建 ##

客户端的构建整体思路与服务端类似，主要也是引导类及pipeline的初始化  
**NettyClient引导类**

@Component
    public class NettyClient {
    
        /**
         * 重连频率，单位：秒
         */
        private static final Integer RECONNECT_SECONDS = 20;
        @Value("${netty.server.host}")
        private String serverHost;
        @Value("${netty.server.port}")
        private Integer serverPort;
        /**
         * 线程组，用于客户端对服务端的链接、数据读写
         */
        private EventLoopGroup eventGroup = new NioEventLoopGroup();
        /**
         * Netty Client Channel
         */
        private Channel channel;
    
        private final NettyClientHandlerInitializer nettyClientHandlerInitializer;
    
        public NettyClient(NettyClientHandlerInitializer nettyClientHandlerInitializer) {
            this.nettyClientHandlerInitializer = nettyClientHandlerInitializer;
        }
    
        /**
         * Netty Client启动
         */
        @PostConstruct
        public void start() {
            // 创建引导
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(eventGroup)
                    // 指定 Channel 为客户端 NioSocketChannel
                    .channel(NioSocketChannel.class)
                    // 指定链接服务器的地址
                    .remoteAddress(serverHost, serverPort)
                    // TCP Keepalive 机制，实现 TCP 层级的心跳保活功能
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    // 允许较小的数据包的发送，降低延迟
                    .option(ChannelOption.TCP_NODELAY, true)
                    .handler(nettyClientHandlerInitializer);
            // 连接服务端，并异步等待成功，即启动客户端
            bootstrap.connect().addListener((ChannelFutureListener) future -> {
                // 连接失败
                if (!future.isSuccess()) {
                    log.error("[start][Netty Client 连接服务器({}:{}) 失败]", serverHost, serverPort);
                    reconnect();
                    return;
                }
                // 连接成功
                channel = future.channel();
                log.info("[start][Netty Client 连接服务器({}:{}) 成功]", serverHost, serverPort);
            });
        }
    
        public void reconnect() {
            eventGroup.schedule(new Runnable() {
                @Override
                public void run() {
                    log.info("[reconnect][开始重连]");
                    try {
                        start();
                    } catch (Exception e) {
                        log.error("[reconnect][重连失败]", e);
                    }
                }
            }, RECONNECT_SECONDS, TimeUnit.SECONDS);
            log.info("[reconnect][{} 秒后将发起重连]", RECONNECT_SECONDS);
        }
    
        @PreDestroy
        public void gracefulShutdown() {
            if (Objects.nonNull(channel)) {
                // 关闭Netty Client
                channel.close();
            }
            eventGroup.shutdownGracefully();
        }
    }

对重连进行了一个简单的实现：调用 \#connect() 方法，连接服务器，并异步等待成功，即启动客户端。同时，添加回调监听器 ChannelFutureListener，在连接服务端失败的时候，调用 \#reconnect() 方法，实现定时重连。

public class NettyClientHandlerInitializer extends ChannelInitializer<Channel> {
    
        /**
         * 心跳超时时间
         */
        private static final Integer READ_TIMEOUT_SECONDS = 60;
    
        private MessageDispatcher messageDispatcher;
        @Autowired
        public void setMessageDispatcher(MessageDispatcher messageDispatcher) {
            this.messageDispatcher = messageDispatcher;
        }
    
        private NettyClientHandler nettyClientHandler;
        @Autowired
        public void setNettyClientHandler(NettyClientHandler nettyClientHandler) {
            this.nettyClientHandler = nettyClientHandler;
        }
    
        @Override
        public void initChannel(Channel ch) {
            ch.pipeline()
                    .addLast(new IdleStateHandler(READ_TIMEOUT_SECONDS, 0, 0))
                    .addLast(new ReadTimeoutHandler(3 * READ_TIMEOUT_SECONDS))
                    .addLast(new InvocationEncoder())
                    .addLast(new InvocationDecoder())
                    .addLast(messageDispatcher)
                    // 客户端处理器
                    .addLast(nettyClientHandler);
        }
    }

## 通信协议 ##

无法将一个 Java 对象直接丢到 TCP Socket 当中，而是需要将其转换成 byte 字节数组，才能写入到 TCP Socket 中去。即，需要将消息对象通过序列化，转换成 byte 字节数组。同时，在服务端收到 byte 字节数组时，需要将其又转换成 Java 对象，即反序列化。序列化的工具非常多，例如说 Google 提供的 Protobuf，性能高效，且序列化出来的二进制数据较小。Netty 对 Protobuf 进行集成，提供了相应的编解码器。demo中出于方便，使用 JSON 方式进行序列化。即所有消息信息通过一个json的String进行传递即可。

## 消息处理 ##

一个im系统，会收到各种类型功能消息，例如登陆认证消息，心跳消息，聊天消息等等，因此对于不同类型的消息都要有各自的处理逻辑。demo通过管理一个MessageHandler的容器，在使用时，根据消息类型选择对应的处理类。思路类似SpringMVC中DispatcherServlet，根据请求地址、方法等，将请求分发到匹配的Controller的Method方法上

public class MessageDispatcher extends SimpleChannelInboundHandler<Invocation> {
    
        private static final ThreadFactory FUTURE_THREAD_FACTORY = new ThreadFactoryBuilder().setNameFormat("future-pool-%s").build();
        private static final int MAX_QUEUE_SIZE = 100;
        private static final ExecutorService EXECUTOR =
                new ThreadPoolExecutor(
                        100,
                        100,
                        0,
                        TimeUnit.MILLISECONDS,
                        new LinkedBlockingDeque<>(MAX_QUEUE_SIZE),
                        FUTURE_THREAD_FACTORY);
    
        private MessageHandlerHolder messageHandlerHolder;
    
        @Autowired
        public void setMessageHandlerHolder(MessageHandlerHolder messageHandlerHolder) {
            this.messageHandlerHolder = messageHandlerHolder;
        }
    
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, Invocation msg) {
            MessageHandler messageHandler = messageHandlerHolder.getMessageHandler(msg.getType());
            Class<? extends Message> msgClass = MessageHandlerHolder.getMessageClass(messageHandler);
            Message message = JsonUtils.fromJson(msg.getMessage(), msgClass);
            // 不要在Netty的I/O线程上执行任何非CPU限定的代码，会影响服务器的吞吐量，一般切换到另一个不同的线程
            EXECUTOR.submit(() -> {
                // noinspection unchecked
                messageHandler.execute(ctx.channel(), message);
            });
        }
    }

其中声明了一个EXECUTOR线程池，用于执行业务逻辑，可能包含RPC调用等。目的是为了不占用Netty的worker线程池。因为一个EventLoop会被多个Channel共享，即多个Channel会共享一个线程，假如IO处理，例如说进行数据库的读取。这样，就会导致一个Channel在执行MessageHandler的过程中，阻塞了共享当前线程的其它Channel的数据读取。

## 空闲检测、心跳机制及重连机制 ##

**断开重连时间点：**

1.Netty 客户端启动时，无法连接 Netty 服务端时，发起重连（实现可以参考NettyClient中的reconnect方法）
    2.Netty 客户端运行时，和 Netty 断开连接时，发起重连（可以通过实现channelInactive方法）
        /**
         * 当你的链接断开的时候，这个方法会被netty框架异步调用
         */
        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            log.info("Lost the connection with the server.");
            final EventLoop eventLoop = ctx.channel().eventLoop();
            eventLoop.schedule(() -> {
                log.info("client reconnecting ...");
                nettyClient.reconnect();
            }, 20, TimeUnit.SECONDS);
            // 继续触发事件,即fireChannel方法将消息在pipeline中传递下去，传递至下一个处理器
            super.channelInactive(ctx);
        }

**心跳机制**  
在Netty中，提供IdleStateHandler类，可以实现对三种心跳的检测，分别是

1.  readerIdleTime(为读超时时间,即测试端一定时间内未接受到被测试端消息)
2.  writerIdleTime(为写超时时间即测试端一定时间内向被测试端发送消息)
3.  allIdleTime(所有类型的超时时间)

在Channel的读或者写空闲时间太长时，将会触发一个IdleStateEvent事件，在NettyClientHandler处理器中，在接收到IdleStateEvent事件时(userEventTriggered方法)，客户端向客户端发送一次心跳消息。

@Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            // 空闲时，向服务端发起一次心跳，即心跳机制
            if (evt instanceof IdleStateEvent) {
                log.info("[userEventTriggered][发起一次心跳]");
                HeartbeatRequest heartbeatRequest = new HeartbeatRequest();
                ctx.writeAndFlush(new Invocation(HeartbeatRequest.TYPE, heartbeatRequest))
                        // A ChannelFutureListener that closes the Channel when the operation ended up with a failure or cancellation rather than a success.
                        .addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
            } else {
                super.userEventTriggered(ctx, evt);
            }
        }

**空闲检测**  
可以使用Netty自带的ReadTimeoutHandler设置timeoutSeconds，当超过timeoutSeconds时间没有收到消息，会抛出ReadTimeoutException异常，并关闭channel

## 应用场景举例：注册认证，单聊及群聊简单实现 ##

对于用户注册，简单来说就是管理一个Channel实例与用户唯一标识的映射，可以利用AttributeKey进行channel变量的存储。举个例子：用户注册(其他功能可以参考代码NettyChannelManager)

public void userRegister(Channel channel, String user) {
            Channel existChannel = channels.get(channel.id());
            if (Objects.isNull(existChannel)) {
                log.error("[userRegister][连接({}) 不存在]", channel.id());
                return;
            }
            channel.attr(CHANNEL_ATTR_KEY_USER).set(user);
            userChannels.putIfAbsent(user, channel);
            log.error("[userRegister][连接({}) 用户({})注册]", channel.id(), user);
        }

其他功能，可以参考MessageController测试类，模拟客户端的相关动作

public class MessageController {
    
        private final NettyClient nettyClient;
    
        public MessageController(NettyClient nettyClient) {
            this.nettyClient = nettyClient;
        }
    
        @PostMapping("/send")
        public BaseResponse sendMessage(String type, String message) {
            if (StringUtils.isEmpty(type) || StringUtils.isEmpty(message)) {
                return BaseResponse.newFailResponse().errorCode(ErrorCode.BAD_PARAMS.getCode()).errorMsg(ErrorCode.BAD_PARAMS.getDesc()).build();
            }
            nettyClient.send(new Invocation(type, message));
            return BaseResponse.newSuccResponse().result("success").build();
        }
    }

本地简单使用postman及多host启动客户端及服务端即可模拟，几个例子协议如下  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L01lZ3VzdGFzX0pKQw_size_16_color_FFFFFF_t_70 1]

（1）模拟用户连接(长连接注册)

type ：AUTH_REQUEST
    message ：{"accessToken":"from"}

（2） 模拟单聊消息发送

type ：SINGLE_CHAT_MESSAGE
    message ： {"fromUser":"from","toUser":"to","msgId":1,"content":"测试发送"}

（2） 模拟群聊消息发送

type ：GROUP_CHAT_MESSAGE
    message ： {"fromUser":"from","msgId":1,"content":"测试发送"}

[github]: https://github.com/MonkeyJJC/netty-im-demo
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L01lZ3VzdGFzX0pKQw_size_16_color_FFFFFF_t_70]: /images/20221022/a575709110d444d4a7decc76a49ced4b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L01lZ3VzdGFzX0pKQw_size_16_color_FFFFFF_t_70 1]: /images/20221022/679d013d97b843ad8f5ba68fd5b08c29.png