写在文章开头

其实Netty基于网络连接声明周期暴露了很多提供用户自实现的API，而本文将基于其中的一个拓展点实现连接可靠性，希望对你有帮助。

Hi，我是 sharkChili ，是个不断在硬核技术上作死的 java coder ，是 CSDN的博客专家 ，也是开源项目 Java Guide 的维护者之一，熟悉 Java 也会一点 Go ，偶尔也会在 C源码 边缘徘徊。写过很多有意思的技术博客，也还在研究并输出技术的路上，希望我的文章对你有帮助，非常欢迎你关注我的公众号： 写代码的SharkChili 。

因为近期收到很多读者的私信，所以也专门创建了一个交流群，感兴趣的读者可以通过上方的公众号获取笔者的联系方式完成好友添加，点击备注 “加群” 即可和笔者和笔者的朋友们进行深入交流。

详解Netty客户端断线重连的设计和实现

Netty生命周期中的channelInactive方法

读过笔者往期文章的读者大体是都知道channelInactive这个回调方法，我们从其注释即可知晓：注册的ChannelHandlerContext 的 Channel现在已经是不活跃即已经不可用的连接，就会调用pipeline上所有的处理器执行其内部实现的channelInactive处理剩余业务：

 /**
     * The {@link Channel} of the {@link ChannelHandlerContext} was registered is now inactive and reached its
     * end of lifetime.
     */
    void channelInactive(ChannelHandlerContext ctx) throws Exception;

实际上channelInactive的执行我们也可以通过源码的方式让读者了解，我们以客户端连接为例，一旦客户端断开连接，客户端的selector就会轮循到连接关闭事件，便会将对应客户端的channel取消并调用channelInactive方法：

从源码角度来NioEventLoop轮询到关闭事件后会直接执行该事件closeOnRead方法，其内部判断连接非open状态则会直接调用close进行连接关闭操作：

protected class NioByteUnsafe extends AbstractNioUnsafe {

        private void closeOnRead(ChannelPipeline pipeline) {
            if (isOpen()) {
               	//......
                } else {
                	//调用close执行关闭连接
                    close(voidPromise());
                }
            }
        }

close逻辑内部最终会定位到客户端的socketchannel执行到AbstractChannel的close方法，其内部会向eventLoop注册一个doDeregister的事件，该事件会将客户端socket注册的读写事件取消，完成后就会调用fireChannelInactive走到channelInactive回调，通知当前客户端netty这个socket的远程连接不再活跃，已经断开了：

对此我们给出上图所示的源码片段，改代码位于AbstractChannel的close方法，其内部核心逻辑就是调用fireChannelInactiveAndDeregister移除客户端socket的读写事件并触发channelInactive的回调通知：

private void close(final ChannelPromise promise, final Throwable cause,
                           final ClosedChannelException closeCause, final boolean notify) {
           //......

          
            if (closeExecutor != null) {
                 //......
            } else {
                 //......
                } else {
                //调用fireChannelInactiveAndDeregister移除断开连接的客户端socket并触发channelInactive回调
                    fireChannelInactiveAndDeregister(wasActive);
                }
            }
        }

fireChannelInactiveAndDeregister内部核心逻辑就是deregister方法，可以看到该方法核心逻辑就是提交给eventLoop一个异步任务，也就是我们上图所说的移除客户端读写事件的方法，方法名是doDeregister，完成该方法调用后就会调用fireChannelInactive方法，告知服务端这个客户端channel连接已断开：

private void deregister(final ChannelPromise promise, final boolean fireChannelInactive) {
           //......
            invokeLater(new Runnable() {
                @Override
                public void run() {
                    try {
                    //移除客户端读写事件
                        doDeregister();
                    } catch (Throwable t) {
                        logger.warn("Unexpected exception occurred while deregistering a channel.", t);
                    } finally {
                    //触发客户端channel的channelInactive回调
                        if (fireChannelInactive) {
                            pipeline.fireChannelInactive();
                        }
                      //......
                    }
                }
            });
        }

对此我们给出doDeregister的逻辑，可以看到其内部拿到eventLoop事件轮询器，通过调用cancel移除当前客户端socket读写事件：

   @Override
    protected void doDeregister() throws Exception {
    //通过selectionKey获取断开连接的客户端读写事件的key，通过cancel移除这些事件
        eventLoop().cancel(selectionKey());
    }

Netty断线重连思路与实现

由此我们知晓要想实现断线重连，客户端可以通过重写channelInactive方法，确保在感知到连接断开时再次提交一个连接的延迟事件，知道断线的连接再次恢复，由此保证客户端连接可靠性：

最终我们给出断线重连的ReconnectHandler，其内部逻辑很简单，延迟5秒后向eventLoop提交一个断线重连的异步连接任务直到成功，完成后我们将这个处理器添加到客户端的pipeline即可：

public class ReconnectHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {

		//提交断线重连的延迟任务
        scheduledDoReConnect(ctx);
        
        ctx.fireChannelInactive();
    }


    private ScheduledFuture<?> scheduledDoReConnect(ChannelHandlerContext ctx) {
        //拿到当前channel的eventLoop提交一个连接远程服务端的延迟任务
        ScheduledFuture<?> scheduledFuture = ctx.channel().eventLoop().schedule(() -> {
            ChannelFuture channelFuture = ctx.channel().connect(new InetSocketAddress("127.0.0.1", 8888));
            channelFuture.addListener(f -> {
                if (!f.isSuccess()) {
                    //如果失败则递归调用scheduledDoReConnect再次尝试
                    scheduledDoReConnect(ctx);
                } else {
                    System.out.println("reconnect success.");
                }
            });

        }, 5, TimeUnit.SECONDS);


        return scheduledFuture;

    }

}

小结

自此我们基于Netty生命周期的源码剖析给出客户端断线重连的设计和落地思路，希望对你有帮助。

我是 sharkchili ，CSDN Java 领域博客专家，mini-redis的作者，我想写一些有意思的东西，希望对你有帮助，如果你想实时收到我写的硬核的文章也欢迎你关注我的公众号： 写代码的SharkChili 。 因为近期收到很多读者的私信，所以也专门创建了一个交流群，感兴趣的读者可以通过上方的公众号获取笔者的联系方式完成好友添加，点击备注 “加群” 即可和笔者和笔者的朋友们进行深入交流。

参考

java netty channelInactive实现断线重连:https://blog.51cto.com/u_16213390/11438584 (opens new window)

Netty 断线重连解决方案:https://www.cnblogs.com/wujinsen/p/8949299.html (opens new window)

epoll 使用详解--epoll_ctl:https://www.cnblogs.com/mrying/p/16075678.html (opens new window)

《Netty实战》