从零单排使用-Netty-构建-IM-聊天室

在《芋道 Spring Boot WebSocket 入门》文章中，咱们应用 WebSocket 实现了一个简略的 IM 性能，反对身份认证、私聊音讯、群聊音讯。

而后就有胖友私信艿艿，心愿应用纯 Netty 实现一个相似的性能。良心的艿艿，当然不会给她发红人卡，因而就有了本文。可能有胖友不晓得 Netty 是什么，这里简略介绍下：

“

Netty 是一个 Java 开源框架。

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

也就是说，Netty 是一个基于 NIO 的客户、服务器端编程框架，应用 Netty 能够确保你疾速和简略的开发出一个网络应用，例如实现了某种协定的客户，服务端利用。

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

上面，咱们来新建三个我的项目，如下图所示：

三个我的项目

lab-67-netty-demo-server 我的项目：搭建 Netty 服务端。
lab-67-netty-demo-client 我的项目：搭建 Netty 客户端。
lab-67-netty-demo-common 我的项目：提供 Netty 的根底封装，提供音讯的编解码、散发的性能。

另外，咱们也会提供 Netty 罕用性能的示例：

心跳机制，实现服务端对客户端的存活检测。
断线重连，实现客户端对服务端的从新连贯。

不哔哔，间接开干。

“

情谊提醒：可能会胖友放心，没有 Netty 根底是不是无奈浏览本文？！

艿艿的想法，看！就硬看，依照代码先本人能搭建一下哈~ 文末，艿艿会提供一波 Netty 根底入门的文章。

本大节，咱们先来应用 Netty 构建服务端与客户端的外围代码，让胖友对我的项目的代码有个初始的认知。

“

本文在提供残缺代码示例，可见 https://github.com/YunaiV/SpringBoot-Labs 的 lab-67 目录。

原创不易，给点个 Star 嘿，一起冲鸭！

创立 lab-67-netty-demo-server 我的项目，搭建 Netty 服务端。如下图所示：

我的项目构造

上面，咱们只会临时看看 server 包下的代码，防止信息量过大，击穿胖友的秃头。

创立 NettyServer 类，Netty 服务端。代码如下：

`@Component
public class NettyServer {

private Logger logger = LoggerFactory.getLogger(getClass());

@Value(“${netty.port}”)
private Integer port;

@Autowired
private NettyServerHandlerInitializer nettyServerHandlerInitializer;

    /**
     * boss 线程组，用于服务端承受客户端的连贯
     */
    private EventLoopGroup bossGroup = new NioEventLoopGroup();
    /**
     * worker 线程组，用于服务端承受客户端的数据读写
     */
    private EventLoopGroup workerGroup = new NioEventLoopGroup();
    /**
     * Netty Server Channel
     */
    private Channel channel;

    /**
     * 启动 Netty Server
     */
    @PostConstruct
    public void start() throws InterruptedException {
        // <2.1> 创立 ServerBootstrap 对象，用于 Netty Server 启动
        ServerBootstrap bootstrap = new ServerBootstrap();
        // <2.2> 设置 ServerBootstrap 的各种属性
        bootstrap.group(bossGroup, workerGroup) // <2.2.1> 设置两个 EventLoopGroup 对象
                .channel(NioServerSocketChannel.class)  // <2.2.2> 指定 Channel 为服务端 NioServerSocketChannel
                .localAddress(new InetSocketAddress(port)) // <2.2.3> 设置 Netty Server 的端口
                .option(ChannelOption.SO_BACKLOG, 1024) // <2.2.4> 服务端 accept 队列的大小
                .childOption(ChannelOption.SO_KEEPALIVE, true) // <2.2.5> TCP Keepalive 机制，实现 TCP 层级的心跳保活性能
                .childOption(ChannelOption.TCP_NODELAY, true) // <2.2.6> 容许较小的数据包的发送，升高提早
                .childHandler(nettyServerHandlerInitializer);
        // <2> 绑定端口，并同步期待胜利，即启动服务端
        ChannelFuture future = bootstrap.bind().sync();
        if (future.isSuccess()) {
            channel = future.channel();
            logger.info(“start”, port);
        }
    }

    /**
     * 敞开 Netty Server
     */
    @PreDestroy
    public void shutdown() {
        // <3.1> 敞开 Netty Server
        if (channel != null) {
            channel.close();
        }
        // <3.2> 优雅敞开两个 EventLoopGroup 对象
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
    }

}
`

???? ① 在类上，增加 @Component 注解，把 NettyServer 的创立交给 Spring 治理。

port 属性，读取 application.yml 配置文件的 netty.port 配置项。
#start() 办法，增加 @PostConstruct 注解，启动 Netty 服务器。
#shutdown() 办法，增加 @PreDestroy 注解，敞开 Netty 服务器。

???? ② 咱们来具体看看 #start() 办法的代码，如何实现 Netty Server 的启动。

<2.1> 处，创立 ServerBootstrap 类，Netty 提供的 服务器 的启动类，不便咱们初始化 Server。

<2.2> 处，设置 ServerBootstrap 的各种属性。

“

情谊提醒：这里波及较多 Netty 组件的常识，艿艿先以简略的语言形容，后续胖友在文末的 Netty 根底入门的文章，补充学噢。

<2.2.1> 处，调用 #group(EventLoopGroup parentGroup, EventLoopGroup childGroup) 办法，设置应用 bossGroup 和 workerGroup。其中：

bossGroup 属性：Boss 线程组，用于服务端承受客户端的连贯。
workerGroup 属性：Worker 线程组，用于服务端承受客户端的 数据读写。

“

Netty 采纳的是多 Reactor 多线程的模型，服务端能够承受更多客户端的数据读写的能力。起因是：

创立专门用于承受 客户端连贯 的 bossGroup 线程组，防止因为已连贯的客户端的数据读写频繁，影响新的客户端的连贯。

创立专门用于接管 客户端读写 的 workerGroup 线程组，多个线程进行客户端的数据读写，能够反对更多客户端。

课后习题：感兴趣的胖友，后续能够看看《【NIO 系列】——之 Reactor 模型》文章。

<2.2.2> 处，调用 #channel(Class<? extends C> channelClass) 办法，设置应用 NioServerSocketChannel 类，它是 Netty 定义的 NIO 服务端 TCP Socket 实现类。

<2.2.3> 处，调用 #localAddress(SocketAddress localAddress) 办法，设置服务端的端口。

<2.2.4> 处，调用 option#(ChannelOption<T> option, T value) 办法，设置服务端承受客户端的 连贯队列 大小。因为 TCP 建设连贯是三次握手，所以第一次握手实现后，会增加到服务端的连贯队列中。

“

课后习题：更多相干内容，后续能够看看《浅谈 TCP Socket 的 backlog 参数》文章。

<2.2.5> 处，调用 #childOption(ChannelOption<T> childOption, T value) 办法，TCP Keepalive 机制，实现 TCP 层级的 心跳保活 性能。

“

课后习题：更多相干内容，后续能够看看《TCP Keepalive 机制刨根问底》文章。

<2.2.6> 处，调用 #childOption(ChannelOption<T> childOption, T value) 办法，容许 较小的数据包 的发送，升高提早。

“

课后习题：更多相干内容，后续能够看看《详解 Socket 编程 — TCP_NODELAY 选项》文章。

<2.2.7> 处，调用 #childHandler(ChannelHandler childHandler) 办法，设置客户端连贯上来的 Channel 的处理器为 NettyServerHandlerInitializer。稍后咱们在「2.1.2 NettyServerHandlerInitializer」大节来看看。

<2.3> 处，调用 #bind() + #sync() 办法，绑定端口，并同步期待胜利，即启动服务端。

???? ③ 咱们来具体看看 #shutdown() 办法的代码，如何实现 Netty Server 的敞开。

<3.1> 处，调用 Channel 的 #close() 办法，敞开 Netty Server，这样客户端就不再能连贯了。

<3.2> 处，调用 EventLoopGroup 的 #shutdownGracefully() 办法，优雅敞开 EventLoopGroup。例如说，它们外面的线程池。

在看 NettyServerHandlerInitializer 的代码之前，咱们须要先理解下 Netty 的 ChannelHandler 组件，用来解决 Channel 的各种事件。这里的事件很宽泛，比方能够是连贯、数据读写、异样、数据转换等等。

ChannelHandler 有十分多的子类，其中有个十分非凡的 ChannelInitializer，它用于 Channel 创立时，实现自定义的初始化逻辑。这里咱们创立的 NettyServerHandlerInitializer 类，就继承了 ChannelInitializer 抽象类，代码如下：

`@Component
public class NettyServerHandlerInitializer extends ChannelInitializer<Channel> {

    /**
     * 心跳超时工夫
     */
    private static final Integer READ_TIMEOUT_SECONDS = 3 * 60;

    @Autowired
    private MessageDispatcher messageDispatcher;
    @Autowired
    private NettyServerHandler nettyServerHandler;

    @Override
    protected void initChannel(Channel ch) {
        // <1> 取得 Channel 对应的 ChannelPipeline
        ChannelPipeline channelPipeline = ch.pipeline();
        // <2> 增加一堆 NettyServerHandler 到 ChannelPipeline 中
        channelPipeline
                // 闲暇检测
                .addLast(new ReadTimeoutHandler(READ_TIMEOUT_SECONDS, TimeUnit.SECONDS))
                // 编码器
                .addLast(new InvocationEncoder())
                // 解码器
                .addLast(new InvocationDecoder())
                // 音讯散发器
                .addLast(messageDispatcher)
                // 服务端处理器
                .addLast(nettyServerHandler)
        ;
    }

}
`

在每一个客户端与服务端建设实现连贯时，服务端会创立一个 Channel 与之对应。此时，NettyServerHandlerInitializer 会进行执行 #initChannel(Channel c) 办法，进行自定义的初始化。

“

情谊提醒：创立的客户端的 Channel，不要和「2.1.1 NettyServer」大节的 NioServerSocketChannel 混同，不是同一个哈。

在 #initChannel(Channel ch) 办法的 ch 参数，就是此时创立的客户端 Channel。

① <1> 处，调用 Channel 的 #pipeline() 办法，取得客户端 Channel 对应的 ChannelPipeline。ChannelPipeline 由一系列的 ChannelHandler 组成，又或者说是 ChannelHandler 链。这样，Channel 所有上所有的事件都会通过 ChannelPipeline，被其上的 ChannelHandler 所解决。

② <2> 处，增加五个 ChannelHandler 到 ChannelPipeline 中，每一个的作用看其上的正文。具体的，咱们会在后续的大节具体解释。

创立 NettyServerHandler 类，继承 ChannelInboundHandlerAdapter 类，实现客户端 Channel 建设连贯、断开连贯、异样时的解决。代码如下：

`@Component
@ChannelHandler.Sharable
public class NettyServerHandler extends ChannelInboundHandlerAdapter {

private Logger logger = LoggerFactory.getLogger(getClass());

@Autowired
private NettyChannelManager channelManager;

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        // 从管理器中增加
        channelManager.add(ctx.channel());
    }

    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) {
        // 从管理器中移除
        channelManager.remove(ctx.channel());
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        logger.error(“exceptionCaught”, ctx.channel().id(), cause);
        // 断开连接
        ctx.channel().close();
    }

}
`

① 在类上增加 @ChannelHandler.Sharable 注解，标记这个 ChannelHandler 能够被多个 Channel 应用。

② channelManager 属性，是咱们实现的客户端 Channel 的管理器。

#channelActive(ChannelHandlerContext ctx) 办法，在客户端和服务端建设连贯实现时，调用 NettyChannelManager 的 #add(Channel channel) 办法，增加到其中。
#channelUnregistered(ChannelHandlerContext ctx) 办法，在客户端和服务端断开连贯时，调用 NettyChannelManager 的 #add(Channel channel) 办法，从其中移除。

具体的 NettyChannelManager 的源码，咱们在「2.1.4 NettyChannelManager」大节中来瞅瞅~

③ #exceptionCaught(ChannelHandlerContext ctx, Throwable cause) 办法，在解决 Channel 的事件产生异样时，调用 Channel 的 #close() 办法，断开和客户端的连贯。

创立 NettyChannelManager 类，提供两种性能。

???? ① 客户端 Channel 的治理。代码如下：

`@Component
public class NettyChannelManager {

    /**
     * {@link Channel#attr(AttributeKey)} 属性中，示意 Channel 对应的用户
     */
    private static final AttributeKey<String> CHANNEL_ATTR_KEY_USER = AttributeKey.newInstance(“user”);

private Logger logger = LoggerFactory.getLogger(getClass());

    /**
     * Channel 映射
     */
    private ConcurrentMap<ChannelId, Channel> channels = new ConcurrentHashMap<>();
    /**
     * 用户与 Channel 的映射。
     *
     * 通过它，能够获取用户对应的 Channel。这样，咱们能够向指定用户发送音讯。
     */
    private ConcurrentMap<String, Channel> userChannels = new ConcurrentHashMap<>();

    /**
     * 增加 Channel 到 {@link #channels} 中
     *
     * @param channel Channel
     */
    public void add(Channel channel) {
        channels.put(channel.id(), channel);
        logger.info(“add”, channel.id());
    }

    /**
     * 增加指定用户到 {@link #userChannels} 中
     *
     * @param channel Channel
     * @param user 用户
     */
    public void addUser(Channel channel, String user) {
        Channel existChannel = channels.get(channel.id());
        if (existChannel == null) {
            logger.error(“addUser”, channel.id());
            return;
        }
        // 设置属性
        channel.attr(CHANNEL_ATTR_KEY_USER).set(user);
        // 增加到 userChannels
        userChannels.put(user, channel);
    }

    /**
     * 将 Channel 从 {@link #channels} 和 {@link #userChannels} 中移除
     *
     * @param channel Channel
     */
    public void remove(Channel channel) {
        // 移除 channels
        channels.remove(channel.id());
        // 移除 userChannels
        if (channel.hasAttr(CHANNEL_ATTR_KEY_USER)) {
            userChannels.remove(channel.attr(CHANNEL_ATTR_KEY_USER).get());
        }
        logger.info(“remove”, channel.id());
    }
}
`

???? ② 向客户端 Channel 发送音讯。代码如下：

`@Component
public class NettyChannelManager {

    /**
     * 向指定用户发送音讯
     *
     * @param user 用户
     * @param invocation 音讯体
     */
    public void send(String user, Invocation invocation) {
        // 取得用户对应的 Channel
        Channel channel = userChannels.get(user);
        if (channel == null) {
            logger.error(“send”);
            return;
        }
        if (!channel.isActive()) {
            logger.error(“send”, channel.id());
            return;
        }
        // 发送音讯
        channel.writeAndFlush(invocation);
    }

    /**
     * 向所有用户发送音讯
     *
     * @param invocation 音讯体
     */
    public void sendAll(Invocation invocation) {
        for (Channel channel : channels.values()) {
            if (!channel.isActive()) {
                logger.error(“send”, channel.id());
                return;
            }
            // 发送音讯
            channel.writeAndFlush(invocation);
        }
    }

}
`

创立 pom.xml 文件，引入 Netty 依赖。

`<?xml version=”1.0″ encoding=”UTF-8″?>
<project xmlns=”http://maven.apache.org/POM/4.0.0″
         xmlns:xsi=”http://www.w3.org/2001/XMLSchema-instance”
         xsi:schemaLocation=”http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/m…d”>
    <parent>
        <artifactId>lab-67-netty-demo</artifactId>
        <groupId>cn.iocoder.springboot.labs</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

<artifactId>lab-67-netty-demo-server</artifactId>

    <properties>
        <!– 依赖相干配置 –>
        <spring.boot.version>2.2.4.RELEASE</spring.boot.version>
        <!– 插件相干配置 –>
        <maven.compiler.target>1.8</maven.compiler.target>
        <maven.compiler.source>1.8</maven.compiler.source>
    </properties>

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-parent</artifactId>
                <version>${spring.boot.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

    <dependencies>
        <!– Spring Boot 根底依赖 –>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>

        <!– Netty 依赖 –>
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.50.Final</version>
        </dependency>

        <!– 引入 netty-demo-common 封装 –>
        <dependency>
            <groupId>cn.iocoder.springboot.labs</groupId>
            <artifactId>lab-67-netty-demo-common</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>
    </dependencies>

</project>
`

创立 NettyServerApplication 类，Netty Server 启动类。代码如下：

`@SpringBootApplication
public class NettyServerApplication {

    public static void main(String[] args) {
        SpringApplication.run(NettyServerApplication.class, args);
    }

}
`

执行 NettyServerApplication 类，启动 Netty Server 服务器。日志如下：

`… // 省略其余日志

2020-06-21 00:16:38.801 INFO 41948 — [main] c.i.s.l.n.server.NettyServer : start
2020-06-21 00:16:38.893 INFO 41948 — [main] c.i.s.l.n.NettyServerApplication : Started NettyServerApplication in 0.96 seconds (JVM running for 1.4)
`

Netty Server 启动在 8888 端口。

创立 lab-67-netty-demo-client 我的项目，搭建 Netty 客户端。如下图所示：

我的项目构造

上面，咱们只会临时看看 client 包下的代码，防止信息量过大，击穿胖友的秃头。

创立 NettyClient 类，Netty 客户端。代码如下：

`@Component
public class NettyClient {

    /**
     * 重连频率，单位：秒
     */
    private static final Integer RECONNECT_SECONDS = 20;

private Logger logger = LoggerFactory.getLogger(getClass());

    @Value(“${netty.server.host}”)
    private String serverHost;
    @Value(“${netty.server.port}”)
    private Integer serverPort;

@Autowired
private NettyClientHandlerInitializer nettyClientHandlerInitializer;

    /**
     * 线程组，用于客户端对服务端的连贯、数据读写
     */
    private EventLoopGroup eventGroup = new NioEventLoopGroup();
    /**
     * Netty Client Channel
     */
    private volatile Channel channel;

    /**
     * 启动 Netty Server
     */
    @PostConstruct
    public void start() throws InterruptedException {
        // <2.1> 创立 Bootstrap 对象，用于 Netty Client 启动
        Bootstrap bootstrap = new Bootstrap();
        // <2.2>
        bootstrap.group(eventGroup) // <2.2.1> 设置一个 EventLoopGroup 对象
                .channel(NioSocketChannel.class)  // <2.2.2> 指定 Channel 为客户端 NioSocketChannel
                .remoteAddress(serverHost, serverPort) // <2.2.3> 指定连贯服务器的地址
                .option(ChannelOption.SO_KEEPALIVE, true) // <2.2.4> TCP Keepalive 机制，实现 TCP 层级的心跳保活性能
                .option(ChannelOption.TCP_NODELAY, true) //<2.2.5>  容许较小的数据包的发送，升高提早
                .handler(nettyClientHandlerInitializer);
        // <2.3> 连贯服务器，并异步期待胜利，即启动客户端
        bootstrap.connect().addListener(new ChannelFutureListener() {

            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                // 连贯失败
                if (!future.isSuccess()) {
                    logger.error(“start”, serverHost, serverPort);
                    reconnect();
                    return;
                }
                // 连贯胜利
                channel = future.channel();
                logger.info(“start”, serverHost, serverPort);
            }

});
}

    public void reconnect() {
        // … 临时省略代码。
    }

    /**
     * 敞开 Netty Server
     */
    @PreDestroy
    public void shutdown() {
        // <3.1> 敞开 Netty Client
        if (channel != null) {
            channel.close();
        }
        // <3.2> 优雅敞开一个 EventLoopGroup 对象
        eventGroup.shutdownGracefully();
    }

    /**
     * 发送音讯
     *
     * @param invocation 音讯体
     */
    public void send(Invocation invocation) {
        if (channel == null) {
            logger.error(“send”);
            return;
        }
        if (!channel.isActive()) {
            logger.error(“send”, channel.id());
            return;
        }
        // 发送音讯
        channel.writeAndFlush(invocation);
    }

}
`

“

情谊提醒：整体代码，是和「2.1.1 NettyServer」对等，且根本是统一的。

???? ① 在类上，增加 @Component 注解，把 NettyClient 的创立交给 Spring 治理。

serverHost 和 serverPort 属性，读取 application.yml 配置文件的 netty.server.host 和 netty.server.port 配置项。
#start() 办法，增加 @PostConstruct 注解，启动 Netty 客户端。
#shutdown() 办法，增加 @PreDestroy 注解，敞开 Netty 客户端。

???? ② 咱们来具体看看 #start() 办法的代码，如何实现 Netty Client 的启动，建设和服务器的连贯。

<2.1> 处，创立 Bootstrap 类，Netty 提供的 客户端 的启动类，不便咱们初始化 Client。

<2.2> 处，设置 Bootstrap 的各种属性。

<2.2.1> 处，调用 #group(EventLoopGroup group) 办法，设置应用 eventGroup 线程组，实现客户端对服务端的连贯、数据读写。

<2.2.2> 处，调用 #channel(Class<? extends C> channelClass) 办法，设置应用 NioSocketChannel 类，它是 Netty 定义的 NIO 服务端 TCP Client 实现类。

<2.2.3> 处，调用 #remoteAddress(SocketAddress localAddress) 办法，设置连贯服务端的地址。

<2.2.4> 处，调用 #option(ChannelOption<T> childOption, T value) 办法，TCP Keepalive 机制，实现 TCP 层级的 心跳保活 性能。

<2.2.5> 处，调用 #childOption(ChannelOption<T> childOption, T value) 办法，容许 较小的数据包 的发送，升高提早。

<2.2.7> 处，调用 #handler(ChannelHandler childHandler) 办法，设置本人 Channel 的处理器为 NettyClientHandlerInitializer。稍后咱们在「2.2.2 NettyClientHandlerInitializer」大节来看看。

<2.3> 处，调用 #connect() 办法，连贯服务器，并异步期待胜利，即启动客户端。同时，增加回调监听器 ChannelFutureListener，在连贯服务端失败的时候，调用 #reconnect() 办法，实现定时重连。???? 具体 #reconnect() 办法的代码，咱们稍后在瞅瞅哈。

③ 咱们来具体看看 #shutdown() 办法的代码，如何实现 Netty Client 的敞开。

<3.1> 处，调用 Channel 的 #close() 办法，敞开 Netty Client，这样客户端就断开和服务端的连贯。

<3.2> 处，调用 EventLoopGroup 的 #shutdownGracefully() 办法，优雅敞开 EventLoopGroup。例如说，它们外面的线程池。

④ #send(Invocation invocation) 办法，实现向服务端发送音讯。

因为 NettyClient 是客户端，所以无需像 NettyServer 一样应用「2.1.4 NettyChannelManager」保护 Channel 的汇合。

创立的 NettyClientHandlerInitializer 类，就继承了 ChannelInitializer 抽象类，实现和服务端建设连贯后，增加相应的 ChannelHandler 处理器。代码如下：

`@Component
public class NettyClientHandlerInitializer extends ChannelInitializer<Channel> {

    /**
     * 心跳超时工夫
     */
    private static final Integer READ_TIMEOUT_SECONDS = 60;

@Autowired
private MessageDispatcher messageDispatcher;

@Autowired
private NettyClientHandler nettyClientHandler;

    @Override
    protected 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)
        ;
    }

}
`

和「2.1.2 NettyServerHandlerInitializer」的代码根本一样，差异在于闲暇检测额定减少 IdleStateHandler，客户端处理器换成了 NettyClientHandler。

创立 NettyClientHandler 类，实现客户端 Channel 断开连贯、异样时的解决。代码如下：

`@Component
@ChannelHandler.Sharable
public class NettyClientHandler extends ChannelInboundHandlerAdapter {

private Logger logger = LoggerFactory.getLogger(getClass());

@Autowired
private NettyClient nettyClient;

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        // 发动重连
        nettyClient.reconnect();
        // 持续触发事件
        super.channelInactive(ctx);
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object event) throws Exception {
        // 闲暇时，向服务端发动一次心跳
        if (event instanceof IdleStateEvent) {
            logger.info(“userEventTriggered”);
            HeartbeatRequest heartbeatRequest = new HeartbeatRequest();
            ctx.writeAndFlush(new Invocation(HeartbeatRequest.TYPE, heartbeatRequest))
                    .addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
        } else {
            super.userEventTriggered(ctx, event);
        }
    }

}
`

① 在类上增加 @ChannelHandler.Sharable 注解，标记这个 ChannelHandler 能够被多个 Channel 应用。

② #channelInactive(ChannelHandlerContext ctx) 办法，实现在和服务端断开连贯时，调用 NettyClient 的 #reconnect() 办法，实现客户端定时和服务端重连。

③ #exceptionCaught(ChannelHandlerContext ctx, Throwable cause) 办法，在解决 Channel 的事件产生异样时，调用 Channel 的 #close() 办法，断开和客户端的连贯。

④ #userEventTriggered(ChannelHandlerContext ctx, Object event) 办法，在客户端在闲暇时，向服务端发送一次心跳，即 心跳机制。这块的内容，咱们稍后具体讲讲。

创立 pom.xml 文件，引入 Netty 依赖。

<artifactId>lab-67-netty-demo-client</artifactId>

    <dependencies>
        <!– 实现对 Spring MVC 的自动化配置 –>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <!– Netty 依赖 –>
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.50.Final</version>
        </dependency>

</project>
`

创立 NettyClientApplication 类，Netty Client 启动类。代码如下：

`@SpringBootApplication
public class NettyClientApplication {

    public static void main(String[] args) {
        SpringApplication.run(NettyClientApplication.class, args);
    }

}
`

执行 NettyClientApplication 类，启动 Netty Client 客户端。日志如下：

`… // 省略其余日志

2020-06-21 09:06:12.205 INFO 44029 — [ntLoopGroup-2-1] c.i.s.l.n.client.NettyClient : start
`

同时 Netty Server 服务端发现有一个客户端接入，打印如下日志：

`2020-06-21 09:06:12.268 INFO 41948 — [ntLoopGroup-3-1] c.i.s.l.n.server.NettyChannelManager : add
`

至此，咱们曾经构建 Netty 服务端和客户端实现。因为 Netty 提供的 API 十分便当，所以咱们不会像间接应用 NIO 时，须要解决大量底层且细节的代码。

不过，如上的内容仅仅是本文的开胃菜，正片行将开始！美滋滋，持续往下看，奥利给！

在「2. 构建 Netty 服务端与客户端」大节中，咱们实现了客户端和服务端的连贯性能。而本大节，咱们要让它们两可能说上话，即进行 数据的读写。

在日常我的项目的开发中，前端和后端之间采纳 HTTP 作为通信协议，应用 文本内容 进行交互，数据格式个别是 JSON。然而在 TCP 的世界里，咱们须要本人基于 二进制 构建，构建客户端和服务端的通信协议。

咱们以客户端向服务端发送音讯来举个例子，假如客户端要发送一个登录申请，对应的类如下：

`public class AuthRequest {

    / 用户名 /
    private String username;
    / 明码 /
    private String password;

}
`

显然，咱们无奈将一个 Java 对象间接丢到 TCP Socket 当中，而是须要将其转换成 byte 字节数组 ，能力写入到 TCP Socket 中去。即，须要将音讯对象通过 序列化，转换成 byte 字节数组。
同时，在服务端收到 byte 字节数组时，须要将其又转换成 Java 对象，即 反序列化。不然，服务端对着一串 byte 字节解决个毛线？！

“

情谊提醒：服务端向客户端发消息，也是一样的过程哈！

序列化的工具十分多，例如说 Google 提供的 Protobuf，性能高效，且序列化进去的二进制数据较小。Netty 对 Protobuf 进行集成，提供了相应的编解码器。如下图所示：

Netty protobuf 包

然而思考到很多胖友对 Protobuf 并不理解，因为它实现序列化又减少胖友的额定学习老本。因而，艿艿认真一个捉摸，还是采纳 JSON 形式进行序列化。可能胖友会纳闷，JSON 不是将对象转换成 字符串 吗？嘿嘿，咱们再把字符串转换成 byte 字节数组 就能够啦~

上面，咱们新建 lab-67-netty-demo-common 我的项目，并在 codec 包下，实现咱们自定义的通信协议。如下图所示：

我的项目构造

创立 Invocation 类，通信协议的音讯体。代码如下：

`/**
* 通信协议的音讯体
*/
public class Invocation {

    /**
     * 类型
     */
    private String type;
    /**
     * 音讯，JSON 格局
     */
    private String message;

    // 空构造方法
    public Invocation() {
    }

    public Invocation(String type, String message) {
        this.type = type;
        this.message = message;
    }

    public Invocation(String type, Message message) {
        this.type = type;
        this.message = JSON.toJSONString(message);
    }

    // … 省略 setter、getter、toString 办法
}
`

① type 属性，类型，用于匹配对应的音讯处理器。如果类比 HTTP 协定，type 属性相当于申请地址。

② message 属性，音讯内容，应用 JSON 格局。

另外，Message 是咱们定义的音讯接口。代码如下：

`public interface Message {

// … 空，作为标记接口

}
`

在开始看 Invocation 的编解码处理器之前，咱们先理解下粘包与拆包的概念。

“

如果的内容，援用《Netty 解决粘包和拆包问题的四种计划》文章的内容，进行二次编辑。

产生粘包和拆包问题的次要起因是，操作系统在发送 TCP 数据的时候，底层会有一个缓冲区，例如 1024 个字节大小。

如果一次申请发送的数据量 比拟小 ，没达到缓冲区大小，TCP 则会将多个申请合并为同一个申请进行发送，这就造成了粘包问题。

“

例如说，在《详解 Socket 编程 — TCP_NODELAY 选项》文章中咱们能够看到，在敞开 Nagle 算法时，申请不会期待满足缓冲区大小，而是尽快收回，升高提早。
如果一次申请发送的数据量 比拟大 ，超过了缓冲区大小，TCP 就会将其拆分为屡次发送，这就是拆包，也就是将一个大的包拆分为多个小包进行发送。

如下图展现了粘包和拆包的一个示意图，演示了粘包和拆包的三种状况：

示例图

A 和 B 两个包都刚好满足 TCP 缓冲区的大小，或者说其等待时间曾经达到 TCP 期待时长，从而还是应用两个独立的包进行发送。
A 和 B 两次申请间隔时间内较短，并且数据包较小，因此合并为同一个包发送给服务端。
B 包比拟大，因此将其拆分为两个包 B_1 和 B_2 进行发送，而这里因为拆分后的 B_2 比拟小，其又与 A 包合并在一起发送。

对于粘包和拆包问题，常见的解决方案有三种：

???? ① 客户端在发送数据包的时候，每个包都 固定长度。比方 1024 个字节大小，如果客户端发送的数据长度有余 1024 个字节，则通过补充空格的形式补全到指定长度。

这种形式，艿艿临时没有找到采纳这种形式的案例。

???? ② 客户端在每个包的开端应用固定的 分隔符。例如 \r\n，如果一个包被拆分了，则期待下一个包发送过去之后找到其中的 \r\n，而后对其拆分后的头部局部与前一个包的残余局部进行合并，这样就失去了一个残缺的包。

具体的案例，有 HTTP、WebSocket、Redis。

???? ③ 将音讯分为头部和音讯体，在头部中保留有以后整个 音讯的长度，只有在读取到足够长度的音讯之后才算是读到了一个残缺的音讯。

“

情谊提醒：计划 ③ 是 ① 的升级版，动静长度。

本文，艿艿将采纳这种形式，在每次 Invocation 序列化成字节数组写入 TCP Socket 之前，先将字节数组的长度写到其中。如下图所示：

Invocation 序列化

创立 InvocationEncoder 类，实现将 Invocation 序列化，并写入到 TCP Socket 中。代码如下：

`public class InvocationEncoder extends MessageToByteEncoder<Invocation> {

private Logger logger = LoggerFactory.getLogger(getClass());

    @Override
    protected void encode(ChannelHandlerContext ctx, Invocation invocation, ByteBuf out) {
        // <2.1> 将 Invocation 转换成 byte[] 数组
        byte[] content = JSON.toJSONBytes(invocation);
        // <2.2> 写入 length
        out.writeInt(content.length);
        // <2.3> 写入内容
        out.writeBytes(content);
        logger.info(“encode”, ctx.channel().id(), invocation.toString());
    }

}
`

① MessageToByteEncoder 是 Netty 定义的编码 ChannelHandler 抽象类，将泛型 <I> 音讯转换成字节数组。

② #encode(ChannelHandlerContext ctx, Invocation invocation, ByteBuf out) 办法，进行编码的逻辑。

<2.1> 处，调用 JSON 的 #toJSONBytes(Object object, SerializerFeature... features) 办法，将 Invocation 转换成字节数组。

<2.2> 处，将字节数组的长度，写入到 TCP Socket 当中。这样，后续「3.4 InvocationDecoder」能够依据该长度，解析到音讯， 解决粘包和拆包的问题。

“

情谊提醒：MessageToByteEncoder 会最终将 ByteBuf out 写到 TCP Socket 中。

<2.3> 处，将字节数组，写入到 TCP Socket 当中。

创立 InvocationDecoder 类，实现从 TCP Socket 读取字节数组，反序列化成 Invocation。代码如下：

`public class InvocationDecoder extends ByteToMessageDecoder {

private Logger logger = LoggerFactory.getLogger(getClass());

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        // <2.1> 标记以后读取地位
        in.markReaderIndex();
        // <2.2> 判断是否可能读取 length 长度
        if (in.readableBytes() <= 4) {
            return;
        }
        // <2.3> 读取长度
        int length = in.readInt();
        if (length < 0) {
            throw new CorruptedFrameException(“negative length: ” + length);
        }
        // <3.1> 如果 message 不够可读，则退回到原读取地位
        if (in.readableBytes() < length) {
            in.resetReaderIndex();
            return;
        }
        // <3.2> 读取内容
        byte[] content = new byte[length];
        in.readBytes(content);
        // <3.3> 解析成 Invocation
        Invocation invocation = JSON.parseObject(content, Invocation.class);
        out.add(invocation);
        logger.info(“decode”, ctx.channel().id(), invocation.toString());
    }

}
`

① ByteToMessageDecoder 是 Netty 定义的解码 ChannelHandler 抽象类，在 TCP Socket 读取到 新数据 时，触发进行解码。

② 在 <2.1>、<2.2>、<2.3> 处，从 TCP Socket 中读取长度。

③ 在 <3.1>、<3.2>、<3.3> 处，从 TCP Socket 中读取 字节数组，并反序列化成 Invocation 对象。

最终，增加 List<Object> out 中，交给后续的 ChannelHandler 进行解决。稍后，咱们将在「4. 音讯散发」小结中，会看到 MessageDispatcher 将 Invocation 散发到其对应的 MessageHandler 中，进行业务逻辑的执行。

创立 pom.xml 文件，引入 Netty、FastJSON 等等依赖。

<artifactId>lab-67-netty-demo-common</artifactId>

    <dependencies>
        <!– Netty 依赖 –>
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.50.Final</version>
        </dependency>

        <!– FastJSON 依赖 –>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.71</version>
        </dependency>

        <!– 引入 Spring 相干依赖 –>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-aop</artifactId>
            <version>5.2.5.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-context</artifactId>
            <version>5.2.5.RELEASE</version>
        </dependency>

        <!– 引入 SLF4J 依赖 –>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>1.7.30</version>
        </dependency>
    </dependencies>

</project>`

至此，咱们曾经实现通信协议的定义、编解码的逻辑，是不是蛮乏味的？！

另外，咱们在 NettyServerHandlerInitializer 和 NettyClientHandlerInitializer 的初始化代码中，将编解码器增加到其中。如下图所示：

编解码器的初始化

在 SpringMVC 中，DispatcherServlet 会依据申请地址、办法等，将申请散发到匹配的 Controller 的 Method 办法上。

在 lab-67-netty-demo-client 我的项目的 dispatcher 包中，咱们创立了 MessageDispatcher 类，实现和 DispatcherServlet 相似的性能，将 Invocation 散发到其对应的 MessageHandler 中，进行业务逻辑的执行。

dispatcher 包

上面，咱们来看看具体的代码实现。

创立 Message 接口，定义音讯的标记接口。代码如下：

`public interface Message {
}
`

下图，是咱们波及到的 Message 实现类。如下图所示：

Message 实现类

创立 MessageHandler 接口，音讯处理器接口。代码如下：

`public interface MessageHandler<T extends Message> {

    /**
     * 执行解决音讯
     *
     * @param channel 通道
     * @param message 音讯
     */
    void execute(Channel channel, T message);

    /**
     * @return 音讯类型，即每个 Message 实现类上的 TYPE 动态字段
     */
    String getType();

}
`

定义了泛型 <T>，须要是 Message 的实现类。
定义的两个接口办法，胖友本人看下正文哈。

下图，是咱们波及到的 MessageHandler 实现类。如下图所示：

MessageHandler 实现类

创立 MessageHandlerContainer 类，作为 MessageHandler 的容器。代码如下：

`public class MessageHandlerContainer implements InitializingBean {

private Logger logger = LoggerFactory.getLogger(getClass());

    /**
     * 音讯类型与 MessageHandler 的映射
     */
    private final Map<String, MessageHandler> handlers = new HashMap<>();

@Autowired
private ApplicationContext applicationContext;

    @Override
    public void afterPropertiesSet() throws Exception {
        // 通过 ApplicationContext 取得所有 MessageHandler Bean
        applicationContext.getBeansOfType(MessageHandler.class).values() // 取得所有 MessageHandler Bean
                .forEach(messageHandler -> handlers.put(messageHandler.getType(), messageHandler)); // 增加到 handlers 中
        logger.info(“afterPropertiesSet”, handlers.size());
    }

    /**
     * 取得类型对应的 MessageHandler
     *
     * @param type 类型
     * @return MessageHandler
     */
    MessageHandler getMessageHandler(String type) {
        MessageHandler handler = handlers.get(type);
        if (handler == null) {
            throw new IllegalArgumentException(String.format(“ 类型(%s) 找不到匹配的 MessageHandler 处理器 ”, type));
        }
        return handler;
    }

    /**
     * 取得 MessageHandler 解决的音讯类
     *
     * @param handler 处理器
     * @return 音讯类
     */
    static Class<? extends Message> getMessageClass(MessageHandler handler) {
        // 取得 Bean 对应的 Class 类名。因为有可能被 AOP 代理过。
        Class<?> targetClass = AopProxyUtils.ultimateTargetClass(handler);
        // 取得接口的 Type 数组
        Type[] interfaces = targetClass.getGenericInterfaces();
        Class<?> superclass = targetClass.getSuperclass();
        while ((Objects.isNull(interfaces) || 0 == interfaces.length) && Objects.nonNull(superclass)) {// 此处，是以父类的接口为准
            interfaces = superclass.getGenericInterfaces();
            superclass = targetClass.getSuperclass();
        }
        if (Objects.nonNull(interfaces)) {
            // 遍历 interfaces 数组
            for (Type type : interfaces) {
                // 要求 type 是泛型参数
                if (type instanceof ParameterizedType) {
                    ParameterizedType parameterizedType = (ParameterizedType) type;
                    // 要求是 MessageHandler 接口
                    if (Objects.equals(parameterizedType.getRawType(), MessageHandler.class)) {
                        Type[] actualTypeArguments = parameterizedType.getActualTypeArguments();
                        // 取首个元素
                        if (Objects.nonNull(actualTypeArguments) && actualTypeArguments.length > 0) {
                            return (Class<Message>) actualTypeArguments[0];
                        } else {
                            throw new IllegalStateException(String.format(“ 类型(%s) 取得不到音讯类型 ”, handler));
                        }
                    }
                }
            }
        }
        throw new IllegalStateException(String.format(“ 类型(%s) 取得不到音讯类型 ”, handler));
    }

}
`

① 实现 InitializingBean 接口，在 #afterPropertiesSet() 办法中，扫描所有 MessageHandler Bean，增加到 MessageHandler 汇合中。

② 在 #getMessageHandler(String type) 办法中，取得类型对应的 MessageHandler 对象。稍后，咱们会在 MessageDispatcher 调用该办法。

③ 在 #getMessageClass(MessageHandler handler) 办法中，通过 MessageHandler 中，通过解析其类上的泛型，取得音讯类型对应的 Class 类。这是参考 rocketmq-spring 我的项目的 DefaultRocketMQListenerContainer#getMessageType() 办法，进行稍微批改。

“

情谊提醒：如果胖友对 Java 的泛型机制没有做过一点理解，可能稍微有点硬核。能够先临时跳过，晓得用意即可。

创立 MessageDispatcher 类，将 Invocation 散发到其对应的 MessageHandler 中，进行业务逻辑的执行。代码如下：

`@ChannelHandler.Sharable
public class MessageDispatcher extends SimpleChannelInboundHandler<Invocation> {

@Autowired
private MessageHandlerContainer messageHandlerContainer;

private final ExecutorService executor = Executors.newFixedThreadPool(200);

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Invocation invocation) {
        // <3.1> 取得 type 对应的 MessageHandler 处理器
        MessageHandler messageHandler = messageHandlerContainer.getMessageHandler(invocation.getType());
        // 取得  MessageHandler 处理器的音讯类
        Class<? extends Message> messageClass = MessageHandlerContainer.getMessageClass(messageHandler);
        // <3.2> 解析音讯
        Message message = JSON.parseObject(invocation.getMessage(), messageClass);
        // <3.3> 执行逻辑
        executor.submit(new Runnable() {

            @Override
            public void run() {
                // noinspection unchecked
                messageHandler.execute(ctx.channel(), message);
            }

});
}

}
`

① 在类上增加 @ChannelHandler.Sharable 注解，标记这个 ChannelHandler 能够被多个 Channel 应用。

② SimpleChannelInboundHandler 是 Netty 定义的 音讯解决 ChannelHandler 抽象类，解决音讯的类型是 <I> 泛型时。

③ #channelRead0(ChannelHandlerContext ctx, Invocation invocation) 办法，解决音讯，进行散发。

音讯散发

<3.1> 处，调用 MessageHandlerContainer 的 #getMessageHandler(String type) 办法，取得 Invocation 的 type 对应的 MessageHandler 处理器。

而后，调用 MessageHandlerContainer 的 #getMessageClass(messageHandler) 办法，取得 MessageHandler 处理器的 音讯类。

<3.2> 处，调用 JSON 的 # parseObject(String text, Class<T> clazz) 办法，将 Invocation 的 message 解析成 MessageHandler 对应的 音讯对象。

<3.3> 处，丢到线程池中，而后调用 MessageHandler 的 #execute(Channel channel, T message) 办法，执行 业务逻辑。

留神，为什么要丢到 executor 线程池中呢？咱们先来理解下 EventGroup 的线程模型。

“

情谊提醒：在咱们启动 Netty 服务端或者客户端时，都会设置其 EventGroup。

EventGroup 咱们能够先简略了解成一个 线程池 ，并且线程池的大小仅仅是 CPU 数量 * 2。每个 Channel 仅仅会被调配到其中的一个线程上，进行数据的读写。并且，多个 Channel 会共享一个线程，即应用同一个线程进行数据的读写。

那么胖友试着思考下，MessageHandler 的具体逻辑眼帘中，往往会波及到 IO 解决 ，例如说进行数据库的读取。这样，就会导致一个 Channel 在执行 MessageHandler 的过程中，阻塞了共享以后线程的其它 Channel 的数据读取。

因而，咱们在这里创立了 executor 线程池，进行 MessageHandler 的逻辑执行，防止阻塞 Channel 的数据读取。

可能会有胖友说，咱们是不是可能把 EventGroup 的线程池设置大一点，例如说 200 呢？对于长连贯的 Netty 服务端，往往会有 1000 ~ 100000 的 Netty 客户端连贯上来，这样无论设置多大的线程池，都会呈现阻塞数据读取的状况。

“

情谊提醒：executor 线程池，咱们个别称之为业务线程池或者逻辑线程池，顾名思义，就是执行业务逻辑的。

这样的设计形式，目前 Dubbo 等等 RPC 框架，都采纳这种形式。

后续，胖友能够认真浏览下《【NIO 系列】——之 Reactor 模型》文章，进一步了解。

创立 NettyServerConfig 配置类，创立 MessageDispatcher 和 MessageHandlerContainer Bean。代码如下：

`@Configuration
public class NettyServerConfig {

    @Bean
    public MessageDispatcher messageDispatcher() {
        return new MessageDispatcher();
    }

    @Bean
    public MessageHandlerContainer messageHandlerContainer() {
        return new MessageHandlerContainer();
    }

}
`

“

情谊提醒：和「4.5 NettyServerConfig」小结统一。

创立 NettyClientConfig 配置类，创立 MessageDispatcher 和 MessageHandlerContainer Bean。代码如下：

`@Configuration
public class NettyClientConfig {

    @Bean
    public MessageDispatcher messageDispatcher() {
        return new MessageDispatcher();
    }

    @Bean
    public MessageHandlerContainer messageHandlerContainer() {
        return new MessageHandlerContainer();
    }

}
`

后续，咱们将在如下大节，具体演示音讯散发的应用：

「6. 心跳机制与闲暇检测」
「7. 认证逻辑」
「7. 单聊逻辑」
「8. 群聊逻辑」

Netty 客户端须要实现 断开重连 机制，解决各种状况下的断开状况。例如说：

Netty 客户端启动时，Netty 服务端处于挂掉，导致无奈连贯上。
在运行过程中，Netty 服务端挂掉，导致连贯被断开。
任一一端网络抖动，导致连贯异样断开。

具体的代码实现比较简单，只须要在两个中央减少重连机制。

Netty 客户端启动时，无奈连贯 Netty 服务端时，发动重连。
Netty 客户端运行时，和 Netty 断开连接时，发动重连。

思考到重连会存在失败的状况，咱们采纳定时重连的形式，防止占用过多资源。

① 在 NettyClient 中，提供 #reconnect() 办法，实现定时重连的逻辑。代码如下：

`// NettyClient.java

public void reconnect() {
    eventGroup.schedule(new Runnable() {
        @Override
        public void run() {
            logger.info(“reconnect”);
            try {
                start();
            } catch (InterruptedException e) {
                logger.error(“reconnect”, e);
            }
        }
    }, RECONNECT_SECONDS, TimeUnit.SECONDS);
    logger.info(“reconnect”, RECONNECT_SECONDS);
}
`

通过调用 EventLoop 提供的 #schedule(Runnable command, long delay, TimeUnit unit) 办法，实现定时逻辑。而在外部的具体逻辑，调用 NettyClient 的 #start() 办法，发动连贯 Netty 服务端。

又因为 NettyClient 在 #start() 办法在连贯 Netty 服务端失败时，又会调用 #reconnect() 办法，从而再次发动 定时重连。如此循环反复，晓得 Netty 客户端连贯上 Netty 服务端。如下图所示：

NettyClient 重连

② 在 NettyClientHandler 中，实现 #channelInactive(ChannelHandlerContext ctx) 办法，在发现和 Netty 服务端断开时，调用 Netty Client 的 #reconnect() 办法，发动重连。代码如下：

`// NettyClientHandler.java

@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
    // 发动重连
    nettyClient.reconnect();
    // 持续触发事件
    super.channelInactive(ctx);
}
`

① 启动 Netty Client，不要启动 Netty Server，控制台打印日志如下图：

重连失败

能够看到 Netty Client 在连贯失败时，一直发动定时重连。

② 启动 Netty Server，控制台打印如下图：

重连胜利

能够看到 Netty Client 胜利重连上 Netty Server。

在上文中，艿艿举荐胖友浏览《TCP Keepalive 机制刨根问底》文章，咱们能够理解到 TCP 自带的闲暇检测机制，默认是 2 小时。这样的检测机制，从 系统资源 层面上来说是能够承受的。

然而在业务层面，如果 2 小时才发现客户端与服务端的连贯理论曾经断开，会导致两头十分多的音讯失落，影响客户的应用体验。

因而，咱们须要在业务层面，本人实现闲暇检测，保障尽快发现客户端与服务端理论曾经断开的状况。实现逻辑如下：

服务端 发现 180 秒未从 客户端 读取到音讯，被动断开连接。
客户端 发现 180 秒未从 服务端 读取到音讯，被动断开连接。

思考到客户端和服务端之间并不是始终有音讯的交互，所以咱们须要减少 心跳机制：

客户端 每 60 秒向 服务端 发动一次心跳音讯，保障 服务端 能够读取到音讯。
服务端 在收到心跳音讯时，回复 客户端 一条确认音讯，保障 客户端 能够读取到音讯。

“

情谊提醒：

为什么是 180 秒？能够加大或者减小，看本人心愿多快检测到连贯异样。过短的工夫，会导致心跳过于频繁，占用过多资源。

为什么是 60 秒？三次机会，确认是否心跳超时。

尽管听起来有点简单，然而实现起来并不简单哈。

在 NettyServerHandlerInitializer 中，咱们增加了一个 ReadTimeoutHandler 处理器，它在超过指定工夫未从对端读取到数据，会抛出 ReadTimeoutException 异样。如下图所示：

ReadTimeoutHandler

通过这样的形式，实现 服务端 发现 180 秒未从 客户端 读取到音讯，被动断开连接。

“

情谊提醒：和「6.1 服务端的闲暇检测」统一。

在 NettyClientHandlerInitializer 中，咱们增加了一个 ReadTimeoutHandler 处理器，它在超过指定工夫未从对端读取到数据，会抛出 ReadTimeoutException 异样。如下图所示：

ReadTimeoutHandler

通过这样的形式，实现 客户端 发现 180 秒未从 服务端 读取到音讯，被动断开连接。

Netty 提供了 IdleStateHandler 处理器，提供闲暇检测的性能，在 Channel 的 读或者写 闲暇工夫太长时，将会触发一个 IdleStateEvent 事件。

这样，咱们只须要在 NettyClientHandler 处理器中，在接管到 IdleStateEvent 事件时，客户端向客户端发送一次心跳音讯。如下图所示：

客户端心跳

其中，HeartbeatRequest 是心跳申请。

同时，咱们在 服务端 我的项目中，创立了一个 HeartbeatRequestHandler 音讯处理器，在收到客户端的心跳申请时，回复客户端一条确认音讯。代码如下：

`@Component
public class HeartbeatRequestHandler implements MessageHandler<HeartbeatRequest> {

private Logger logger = LoggerFactory.getLogger(getClass());

    @Override
    public void execute(Channel channel, HeartbeatRequest message) {
        logger.info(“execute”, channel.id());
        // 响应心跳
        HeartbeatResponse response = new HeartbeatResponse();
        channel.writeAndFlush(new Invocation(HeartbeatResponse.TYPE, response));
    }

    @Override
    public String getType() {
        return HeartbeatRequest.TYPE;
    }

}
`

其中，HeartbeatResponse 是心跳确认响应

启动 Netty Server 服务端，再启动 Netty Client 客户端，急躁期待 60 秒后，能够看到心跳日志如下：

`// … 客户端
2020-06-22 08:24:47.275  INFO 57005 — [ntLoopGroup-2-1] c.i.s.l.n.c.handler.NettyClientHandler   : userEventTriggered
2020-06-22 08:24:47.335  INFO 57005 — [ntLoopGroup-2-1] c.i.s.l.n.codec.InvocationEncoder        : encode
2020-06-22 08:24:47.408  INFO 57005 — [ntLoopGroup-2-1] c.i.s.l.n.codec.InvocationDecoder        : decode
2020-06-22 08:24:47.409  INFO 57005 — [pool-1-thread-1] c.i.s.l.n.m.h.HeartbeatResponseHandler   : execute

// … 服务端
2020-06-22 08:24:47.388  INFO 56998 — [ntLoopGroup-3-1] c.i.s.l.n.codec.InvocationDecoder        : decode
2020-06-22 08:24:47.390  INFO 56998 — [pool-1-thread-1] c.i.s.l.n.m.h.HeartbeatRequestHandler    : execute
2020-06-22 08:24:47.399  INFO 56998 — [ntLoopGroup-3-1] c.i.s.l.n.codec.InvocationEncoder        : encode
`

“

情谊提醒：从本大节开始，咱们就具体看看业务逻辑的解决示例。

认证的过程，如下图所示：

认证流程

创立 AuthRequest 类，定义用户认证申请。代码如下：

`public class AuthRequest implements Message {

public static final String TYPE = “AUTH_REQUEST”;

    /**
     * 认证 Token
     */
    private String accessToken;

    // … 省略 setter、getter、toString 办法
}
`

这里咱们应用 accessToken 认证令牌进行认证。

因为个别状况下，咱们应用 HTTP 进行登录零碎，而后应用登录后的身份标识（例如说 accessToken 认证令牌），将客户端和以后用户进行认证绑定。

创立 AuthResponse 类，定义用户认证响应。代码如下：

`public class AuthResponse implements Message {

public static final String TYPE = “AUTH_RESPONSE”;

    /**
     * 响应状态码
     */
    private Integer code;
    /**
     * 响应提醒
     */
    private String message;

    // … 省略 setter、getter、toString 办法
}
`

“

服务端 …

创立 AuthRequestHandler 类，为 服务端 解决 客户端 的认证申请。代码如下：

`@Component
public class AuthRequestHandler implements MessageHandler<AuthRequest> {

@Autowired
private NettyChannelManager nettyChannelManager;

    @Override
    public void execute(Channel channel, AuthRequest authRequest) {
        // <1> 如果未传递 accessToken
        if (StringUtils.isEmpty(authRequest.getAccessToken())) {
            AuthResponse authResponse = new AuthResponse().setCode(1).setMessage(“ 认证 accessToken 未传入 ”);
            channel.writeAndFlush(new Invocation(AuthResponse.TYPE, authResponse));
            return;
        }

// <2> … 此处应有一段

        // <3> 将用户和 Channel 绑定
        // 思考到代码简化，咱们先间接应用 accessToken 作为 User
        nettyChannelManager.addUser(channel, authRequest.getAccessToken());

        // <4> 响应认证胜利
        AuthResponse authResponse = new AuthResponse().setCode(0);
        channel.writeAndFlush(new Invocation(AuthResponse.TYPE, authResponse));
    }

    @Override
    public String getType() {
        return AuthRequest.TYPE;
    }

}
`

代码比较简单，胖友看看 <1>、<2>、<3>、<4> 上的正文。

“

客户端 …

创立 AuthResponseHandler 类，为 客户端 解决 服务端 的认证响应。代码如下：

`@Component
public class AuthResponseHandler implements MessageHandler<AuthResponse> {

private Logger logger = LoggerFactory.getLogger(getClass());

    @Override
    public void execute(Channel channel, AuthResponse message) {
        logger.info(“execute”, message);
    }

    @Override
    public String getType() {
        return AuthResponse.TYPE;
    }

}
`

打印个认证后果，不便调试。

“

客户端 …

创立 TestController 类，提供 /test/mock 接口，模仿客户端向服务端发送申请。代码如下：

`@RestController
@RequestMapping(“/test”)
public class TestController {

@Autowired
private NettyClient nettyClient;

    @PostMapping(“/mock”)
    public String mock(String type, String message) {
        // 创立 Invocation 对象
        Invocation invocation = new Invocation(type, message);
        // 发送音讯
        nettyClient.send(invocation);
        return “success”;
    }

}
`

启动 Netty Server 服务端，再启动 Netty Client 客户端，而后应用 Postman 模仿一次认证申请。如下图所示：

Postman 模仿认证申请

同时，能够看到认证胜利的日志如下：

`// 客户端 …
2020-06-22 08:41:12.364  INFO 57583 — [ntLoopGroup-2-1] c.i.s.l.n.codec.InvocationEncoder        : encode
2020-06-22 08:41:12.390  INFO 57583 — [ntLoopGroup-2-1] c.i.s.l.n.codec.InvocationDecoder        : decode
2020-06-22 08:41:12.392  INFO 57583 — [pool-1-thread-1] c.i.s.l.n.m.auth.AuthResponseHandler     : execute

// 服务端 …
2020-06-22 08:41:12.374 INFO 56998 — [ntLoopGroup-3-2] c.i.s.l.n.codec.InvocationDecoder : decode
2020-06-22 08:41:12.379 INFO 56998 — [ntLoopGroup-3-2] c.i.s.l.n.codec.InvocationEncoder : encode
`

私聊的过程，如下图所示：

私聊流程

服务端负责将客户端 A 发送的私聊音讯，转发给客户端 B。

创立 ChatSendToOneRequest 类，发送给指定人的私聊音讯的申请。代码如下：

`public class ChatSendToOneRequest implements Message {

public static final String TYPE = “CHAT_SEND_TO_ONE_REQUEST”;

    /**
     * 发送给的用户
     */
    private String toUser;
    /**
     * 音讯编号
     */
    private String msgId;
    /**
     * 内容
     */
    private String content;

    // … 省略 setter、getter、toString 办法
}
`

创立 ChatSendResponse 类，聊天发送音讯后果的响应。代码如下：

`public class ChatSendResponse implements Message {

public static final String TYPE = “CHAT_SEND_RESPONSE”;

    /**
     * 音讯编号
     */
    private String msgId;
    /**
     * 响应状态码
     */
    private Integer code;
    /**
     * 响应提醒
     */
    private String message;

    // … 省略 setter、getter、toString 办法
}
`

创立 ChatRedirectToUserRequest 类，转发音讯给一个用户的申请。代码如下：

`public class ChatRedirectToUserRequest implements Message {

public static final String TYPE = “CHAT_REDIRECT_TO_USER_REQUEST”;

    /**
     * 音讯编号
     */
    private String msgId;
    /**
     * 内容
     */
    private String content;

    // … 省略 setter、getter、toString 办法
}
`

“

情谊提醒：写完之后，艿艿忽然发现少了一个 fromUser 字段，示意来自谁的音讯。

“

服务端 …

创立 ChatSendToOneHandler 类，为 服务端 解决 客户端 的私聊申请。代码如下：

`@Component
public class ChatSendToOneHandler implements MessageHandler<ChatSendToOneRequest> {

@Autowired
private NettyChannelManager nettyChannelManager;

    @Override
    public void execute(Channel channel, ChatSendToOneRequest message) {
        // <1> 这里，伪装间接胜利
        ChatSendResponse sendResponse = new ChatSendResponse().setMsgId(message.getMsgId()).setCode(0);
        channel.writeAndFlush(new Invocation(ChatSendResponse.TYPE, sendResponse));

        // <2> 创立转发的音讯，发送给指定用户
        ChatRedirectToUserRequest sendToUserRequest = new ChatRedirectToUserRequest().setMsgId(message.getMsgId())
                .setContent(message.getContent());
        nettyChannelManager.send(message.getToUser(), new Invocation(ChatRedirectToUserRequest.TYPE, sendToUserRequest));
    }

    @Override
    public String getType() {
        return ChatSendToOneRequest.TYPE;
    }

}
`

代码比较简单，胖友看看 <1>、<2> 上的正文。

“

客户端 …

创立 ChatSendResponseHandler 类，为 客户端 解决 服务端 的聊天响应。代码如下：

`@Component
public class ChatSendResponseHandler implements MessageHandler<ChatSendResponse> {

private Logger logger = LoggerFactory.getLogger(getClass());

    @Override
    public void execute(Channel channel, ChatSendResponse message) {
        logger.info(“execute”, message);
    }

    @Override
    public String getType() {
        return ChatSendResponse.TYPE;
    }

}
`

打印个聊天发送后果，不便调试。

“

客户端

创立 ChatRedirectToUserRequestHandler 类，为 客户端 解决 服务端 的转发音讯的申请。代码如下：

`@Component
public class ChatRedirectToUserRequestHandler implements MessageHandler<ChatRedirectToUserRequest> {

private Logger logger = LoggerFactory.getLogger(getClass());

    @Override
    public void execute(Channel channel, ChatRedirectToUserRequest message) {
        logger.info(“execute”, message);
    }

    @Override
    public String getType() {
        return ChatRedirectToUserRequest.TYPE;
    }

}
`

打印个聊天 接管音讯，不便调试。

① 启动 Netty Server 服务端。

② 启动 Netty Client 客户端 A。而后应用 Postman 模仿一次认证申请(用户为 yunai)。如下图所示：

Postman 模仿认证申请

③ 启动 Netty Client 客户端 B。留神，须要设置 --server.port 端口为 8081，防止抵触。如下图所示：

IDEA 设置

而后应用 Postman 模仿一次认证申请(用户为 tutou)。如下图所示：

Postman 模仿认证申请

④ 最初应用 Postman 模仿一次 yunai 芋艿给 tutou 土豆发送一次私聊音讯。如下图所示：

Postman 模仿私聊申请

同时，能够看到客户端 A 向客户端 B 发送私聊音讯的日志如下：

`// 客户端 A…（芋艿）
2020-06-22 08:48:09.505  INFO 57583 — [ntLoopGroup-2-1] c.i.s.l.n.codec.InvocationEncoder        : decode
2020-06-22 08:48:09.510  INFO 57583 — [ntLoopGroup-2-1] c.i.s.l.n.codec.InvocationDecoder        : decode
2020-06-22 08:48:09.511  INFO 57583 — [ool-1-thread-69] c.i.s.l.n.m.c.ChatSendResponseHandler    : execute
2020-06-22 08:48:35.148  INFO 57583 — [ntLoopGroup-2-1] c.i.s.l.n.codec.InvocationEncoder        : decode
2020-06-22 08:48:35.150  INFO 57583 — [ntLoopGroup-2-1] c.i.s.l.n.codec.InvocationDecoder        : decode
2020-06-22 08:48:35.150  INFO 57583 — [ool-1-thread-70] c.i.s.l.n.m.c.ChatSendResponseHandler    : execute

// 服务端 …
2020-06-22 08:48:35.149  INFO 56998 — [ntLoopGroup-3-2] c.i.s.l.n.codec.InvocationDecoder        : decode
2020-06-22 08:48:35.149  INFO 56998 — [ntLoopGroup-3-2] c.i.s.l.n.codec.InvocationEncoder        : decode
2020-06-22 08:48:35.149  INFO 56998 — [ntLoopGroup-3-3] c.i.s.l.n.codec.InvocationEncoder        : decode

// 客户端 B…（秃头）
2020-06-22 08:48:18.277  INFO 59613 — [ntLoopGroup-2-1] c.i.s.l.n.c.handler.NettyClientHandler   : userEventTriggered
2020-06-22 08:48:18.278  INFO 59613 — [ntLoopGroup-2-1] c.i.s.l.n.codec.InvocationEncoder        : encode
2020-06-22 08:48:18.280  INFO 59613 — [ntLoopGroup-2-1] c.i.s.l.n.codec.InvocationDecoder        : decode
2020-06-22 08:48:18.281  INFO 59613 — [pool-1-thread-4] c.i.s.l.n.m.h.HeartbeatResponseHandler   : execute
2020-06-22 08:48:35.150  INFO 59613 — [ntLoopGroup-2-1] c.i.s.l.n.codec.InvocationDecoder        : decode
2020-06-22 08:48:35.151  INFO 59613 — [pool-1-thread-5] l.n.m.c.ChatRedirectToUserRequestHandler : execute
`

群聊的过程，如下图所示：

群聊流程

服务端负责将客户端 A 发送的群聊音讯，转发给客户端 A、B、C。

“

情谊提醒：思考到逻辑简洁，艿艿提供的本大节的示例，并不是一个一个群，而是所有人在一个大的群聊中哈~

创立 ChatSendToOneRequest 类，发送给所有人的群聊音讯的申请。代码如下：

`public class ChatSendToAllRequest implements Message {

public static final String TYPE = “CHAT_SEND_TO_ALL_REQUEST”;

    /**
     * 音讯编号
     */
    private String msgId;
    /**
     * 内容
     */
    private String content;

    // … 省略 setter、getter、toString 办法
}
`

“

情谊提醒：如果是正经的群聊，会有一个 groupId 字段，示意群编号。

和「8.2 ChatSendResponse」大节统一。

和「8.3 ChatRedirectToUserRequest」大节统一。

“

服务端 …

创立 ChatSendToAllHandler 类，为 服务端 解决 客户端 的群聊申请。代码如下：

`@Component
public class ChatSendToAllHandler implements MessageHandler<ChatSendToAllRequest> {

@Autowired
private NettyChannelManager nettyChannelManager;

    @Override
    public void execute(Channel channel, ChatSendToAllRequest message) {
        // <1> 这里，伪装间接胜利
        ChatSendResponse sendResponse = new ChatSendResponse().setMsgId(message.getMsgId()).setCode(0);
        channel.writeAndFlush(new Invocation(ChatSendResponse.TYPE, sendResponse));

        // <2> 创立转发的音讯，并播送发送
        ChatRedirectToUserRequest sendToUserRequest = new ChatRedirectToUserRequest().setMsgId(message.getMsgId())
                .setContent(message.getContent());
        nettyChannelManager.sendAll(new Invocation(ChatRedirectToUserRequest.TYPE, sendToUserRequest));
    }

    @Override
    public String getType() {
        return ChatSendToAllRequest.TYPE;
    }

}
`

代码比较简单，胖友看看 <1>、<2> 上的正文。

和「8.5 ChatSendResponseHandler」大节统一。

和「8.6 ChatRedirectToUserRequestHandler」大节统一。

① 启动 Netty Server 服务端。

② 启动 Netty Client 客户端 A。而后应用 Postman 模仿一次认证申请(用户为 yunai)。如下图所示：

Postman 模仿认证申请

③ 启动 Netty Client 客户端 B。留神，须要设置 --server.port 端口为 8081，防止抵触。

IDEA 设置

④ 启动 Netty Client 客户端 C。留神，须要设置 --server.port 端口为 8082，防止抵触。

IDEA 设置

⑤ 最初应用 Postman 模仿一次发送群聊音讯。如下图所示：

Postman 模仿群聊申请

同时，能够看到客户端 A 群发给所有客户端的日志如下：

`// 客户端 A…
2020-06-22 08:55:44.898  INFO 57583 — [ntLoopGroup-2-1] c.i.s.l.n.codec.InvocationEncoder        : decode
2020-06-22 08:55:44.901  INFO 57583 — [ntLoopGroup-2-1] c.i.s.l.n.codec.InvocationDecoder        : decode
2020-06-22 08:55:44.901  INFO 57583 — [ol-1-thread-148] c.i.s.l.n.m.c.ChatSendResponseHandler    : execute
2020-06-22 08:55:44.901  INFO 57583 — [ntLoopGroup-2-1] c.i.s.l.n.codec.InvocationDecoder        : decode
2020-06-22 08:55:44.903  INFO 57583 — [ol-1-thread-149] l.n.m.c.ChatRedirectToUserRequestHandler : execute

// 服务端 …
2020-06-22 08:55:44.898  INFO 56998 — [ntLoopGroup-3-2] c.i.s.l.n.codec.InvocationDecoder        : decode
2020-06-22 08:55:44.901  INFO 56998 — [ntLoopGroup-3-2] c.i.s.l.n.codec.InvocationEncoder        : decode
2020-06-22 08:55:44.901  INFO 56998 — [ntLoopGroup-3-2] c.i.s.l.n.codec.InvocationEncoder        : decode
2020-06-22 08:55:44.901  INFO 56998 — [ntLoopGroup-3-3] c.i.s.l.n.codec.InvocationEncoder        : decode
2020-06-22 08:55:44.901  INFO 56998 — [ntLoopGroup-3-4] c.i.s.l.n.codec.InvocationEncoder        : decode

// 客户端 B…
2020-06-22 08:55:44.902 INFO 59613 — [ntLoopGroup-2-1] c.i.s.l.n.codec.InvocationDecoder : decode
2020-06-22 08:55:44.902 INFO 59613 — [ool-1-thread-83] l.n.m.c.ChatRedirectToUserRequestHandler : execute

// 客户端 C…
2020-06-22 08:55:44.901 INFO 61597 — [ntLoopGroup-2-1] c.i.s.l.n.codec.InvocationDecoder : decode
2020-06-22 08:55:44.903 INFO 61597 — [ool-1-thread-16] l.n.m.c.ChatRedirectToUserRequestHandler : execute
`

至此，咱们曾经通过 Netty 实现了一个简略的 IM 性能，是不是播种蛮大的，嘿嘿。

上面，良心的艿艿，再来举荐一波文章，嘿嘿。

想要理解 Netty 源码的，能够浏览《Netty 实现原理与源码解析零碎 —— 精品合集》文章。
想要入门 Netty 根底的，能够浏览《Netty Bootstrap（图解）》文章。

等后续，艿艿会在 https://github.com/YunaiV/onemall 开源我的项目中，实现一个绝对残缺的客服性能，哈哈哈~

1. 概述

2. 构建 Netty 服务端与客户端

2.1 构建 Netty 服务端

2.1.1 NettyServer

2.1.2 NettyServerHandlerInitializer

2.1.3 NettyServerHandler

2.1.4 NettyChannelManager

2.1.5 引入依赖

2.1.6 NettyServerApplication

2.1.7 简略测试

2.2 构建 Netty 客户端

2.2.1 NettyClient

2.2.2 NettyClientHandlerInitializer

2.2.3 NettyClientHandler

2.2.4 引入依赖

2.2.5 NettyClientApplication

2.2.6 简略测试

2.3 小结

3. 通信协议

3.1 Invocation

3.2 粘包与拆包

3.2.1 产生起因

3.2.2 解决方案

3.3 InvocationEncoder

3.4 InvocationDecoder

3.5 引入依赖

3.6 小结

4. 音讯散发

4.1 Message

4.2 MessageHandler

4.3 MessageHandlerContainer

4.4 MessageDispatcher

4.5 NettyServerConfig

4.6 NettyClientConfig

4.7 小结

5. 断开重连

5.1 具体代码

5.2 简略测试

6. 心跳机制与闲暇检测

6.1 服务端的闲暇检测

6.2 客户端的闲暇检测

6.3 心跳机制

6.4 简略测试

7. 认证逻辑

7.1 AuthRequest

7.2 AuthResponse

7.3 AuthRequestHandler

7.4 AuthResponseHandler

7.5 TestController

7.6 简略测试

8. 单聊逻辑

8.1 ChatSendToOneRequest

8.2 ChatSendResponse

8.3 ChatRedirectToUserRequest

8.4 ChatSendToOneHandler

8.5 ChatSendResponseHandler

8.6 ChatRedirectToUserRequestHandler

8.7 简略测试

9. 群聊逻辑

9.1 ChatSendToAllRequest

9.2 ChatSendResponse

9.3 ChatRedirectToUserRequest

9.4 ChatSendToAllHandler

9.5 ChatSendResponseHandler

9.6 ChatRedirectToUserRequestHandler

9.7 简略测试

666. 彩蛋