NettyでIOException:Connection reset by peerとjava.nio.channels.ClosedChannelException: null

最近、多くのIOException:Connection reset by peerとClosedChannelException:nullが発見されました
コードを深く見て、いくつかのテストをして、Connection resetがクライアントがchannelが閉じられていることを知らない場合、eventloopのunsafeをトリガーすることを発見しました.read()操作放出
一方、ClosedChannelExceptionは一般的にNettyによって自発的に投げ出されており、AbstractChannelやSSLhandlerではClosedChannel関連のコードが見られます
AbstractChannel 

static final ClosedChannelException CLOSED_CHANNEL_EXCEPTION = new ClosedChannelException();

...

    static {
        CLOSED_CHANNEL_EXCEPTION.setStackTrace(EmptyArrays.EMPTY_STACK_TRACE);
        NOT_YET_CONNECTED_EXCEPTION.setStackTrace(EmptyArrays.EMPTY_STACK_TRACE);
    }

...

@Override
        public void write(Object msg, ChannelPromise promise) {
            ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
            if (outboundBuffer == null) {
                // If the outboundBuffer is null we know the channel was closed and so
                // need to fail the future right away. If it is not null the handling of the rest
                // will be done in flush0()
                // See https://github.com/netty/netty/issues/2362
                safeSetFailure(promise, CLOSED_CHANNEL_EXCEPTION);
                // release message now to prevent resource-leak
                ReferenceCountUtil.release(msg);
                return;
            }
            outboundBuffer.addMessage(msg, promise);
        }

コードの多くの部分では、このClosedChannelExceptionがあります.つまり、channel close以降にwriteメソッドが呼び出されると、writeのfutureがfailureに設定され、causeがClosedChannelExceptionに設定されます.同じSSLhandlerでも似ています.
-----------------
Connection reset by peerに戻ると、この状況をシミュレートするのは簡単です.サーバ側にchannelActiveのときにclose channelのhandlerを設定します.クライアント側ではConnectに成功した直後にリクエストデータを送信するlistenerを書く.次のように
client

    public static void main(String[] args) throws IOException, InterruptedException {
        Bootstrap b = new Bootstrap();
        b.group(new NioEventLoopGroup())
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                    }
                });
        b.connect("localhost", 8090).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (future.isSuccess()) {
                    future.channel().write(Unpooled.buffer().writeBytes("123".getBytes()));
                    future.channel().flush();
                }
            }
        });

server

public class SimpleServer {

    public static void main(String[] args) throws Exception {

        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        ServerBootstrap b = new ServerBootstrap();
        b.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.SO_REUSEADDR, true)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new SimpleServerHandler());
                    }
                });
        b.bind(8090).sync().channel().closeFuture().sync();
    }
}


public class SimpleServerHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ctx.channel().close().sync();
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, final Object msg) throws Exception {
        System.out.println(123);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("inactive");
    }
}

 
この場合connection reset by peer異常をトリガーできるのは、connectが成功するとclientセグメントが先にconnectに成功したlistenerをトリガーするためで、このときserverセグメントはchannelを切断していますが、channelが切断されたイベントもトリガーされます(クライアントreadイベントがトリガーされますが、このreadは-1を返します.-1はchannelが閉鎖されたことを表し、clientのchannelInactiveとchannel activeの状態の変化はこの時に発生しました).しかし、このイベントはconnectに成功したlistenerの後に実行されるので、この時listenerのchannelは自分が切断されたことを知らないので、writeとflushの操作を続け、flushを呼び出すとeventloopがOP_に入ります.READ事件の時read()はconnection reset異常を投げ出す.eventloopコードは次のとおりです.
NioEventLoop

private static void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
        final NioUnsafe unsafe = ch.unsafe();
        if (!k.isValid()) {
            // close the channel if the key is not valid anymore
            unsafe.close(unsafe.voidPromise());
            return;
        }

        try {
            int readyOps = k.readyOps();
            // Also check for readOps of 0 to workaround possible JDK bug which may otherwise lead
            // to a spin loop
            if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
                unsafe.read(); if (!ch.isOpen()) {
                    // Connection already closed - no need to handle write.
                    return;
                }
            }
            if ((readyOps & SelectionKey.OP_WRITE) != 0) {
                // Call forceFlush which will also take care of clear the OP_WRITE once there is nothing left to write
                ch.unsafe().forceFlush();
            }
            if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
                // remove OP_CONNECT as otherwise Selector.select(..) will always return without blocking
                // See https://github.com/netty/netty/issues/924
                int ops = k.interestOps();
                ops &= ~SelectionKey.OP_CONNECT;
                k.interestOps(ops);

                unsafe.finishConnect();
            }
        } catch (CancelledKeyException e) {
            unsafe.close(unsafe.voidPromise());
        }
    }

これがconnection reset by peerの発生の原因です
------------------
ClosedChannelExceptionがどのように生成されたかを見てみると、彼を再現するのも簡単です.まず、クライアントがアクティブに閉じてからClosedChannelExceptionが現れるわけではないことを明確にしなければならない.次に、ClosedChannelExceptionが表示されるクライアントの書き方を2つ見てみましょう.
クライアント1、チャンネルをアクティブに閉じる

public class SimpleClient {

    private static final Logger logger = LoggerFactory.getLogger(SimpleClient.class);

    public static void main(String[] args) throws IOException, InterruptedException {
        Bootstrap b = new Bootstrap();
        b.group(new NioEventLoopGroup())
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                    }
                });
        b.connect("localhost", 8090).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (future.isSuccess()) {
                    future.channel().close();
                    future.channel().write(Unpooled.buffer().writeBytes("123".getBytes())).addListener(new ChannelFutureListener() {
                        @Override
                        public void operationComplete(ChannelFuture future) throws Exception {
                            if (!future.isSuccess()) {
                                logger.error("Error", future.cause());
                            }
                        }
                    });
                    future.channel().flush();
                }
            }
        });
    }
}

 
writeの前にcloseをアクティブに呼び出すと、writeは必ずcloseがclose状態であることを知っていて、最後にwriteは失敗して、futureの中のcauseはClosedChannelExceptionです
--------------------
client 2. サービス側によるClosedChannelException

public class SimpleClient {

    private static final Logger logger = LoggerFactory.getLogger(SimpleClient.class);

    public static void main(String[] args) throws IOException, InterruptedException {
        Bootstrap b = new Bootstrap();
        b.group(new NioEventLoopGroup())
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                    }
                });
        Channel channel = b.connect("localhost", 8090).sync().channel();
        Thread.sleep(3000);
        channel.writeAndFlush(Unpooled.buffer().writeBytes("123".getBytes())).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (!future.isSuccess()) {
                    logger.error("error", future.cause());
                }
            }
        });
    }
}

サービス側

public class SimpleServer {

    public static void main(String[] args) throws Exception {

        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        ServerBootstrap b = new ServerBootstrap();
        b.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.SO_REUSEADDR, true)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new SimpleServerHandler());
                    }
                });
        b.bind(8090).sync().channel().closeFuture().sync();
    }
}

この場合、サービス側はchannelを閉じ、クライアントはsleepを先に閉じ、この間clientのeventLoopはクライアントが閉じた時間、すなわちeventLoopのprocessKeyメソッドがOP_に入るREAD、それからreadは1つ出てきて、最後にclient channelInactiveイベントをトリガーして、sleepが目が覚めると、クライアントはwriteAndFlushを呼び出して、この時クライアントchannelの状態はすでにinactiveになって、だからwriteは失敗して、causeはClosedChannelExceptionです