Netty 应用与原理

public class BioServer {
    public static void main (String[] args) throws IOException {
        // BIO 模型的服务端要为每一个客户端建立一个对应的连接
        ServerSocket serverSocket = new ServerSocket(1145);
        while (true) {
            // 持续接受客户端的连接
            Socket accept = serverSocket.accept();
            // 为每一个客户端连接新开一个线程，执行对应的业务
            new Thread(new ClientService(accept)).start();
        }
    }

    static class ClientService implements Runnable {
        private Socket socket;

        public ClientService (Socket socket) {
            this.socket = socket;
        }

        @Override
        public void run() {
            System.out.println("执行对应的业务操作：" + socket);
        }
    }
}

public class Client {
    public static void main(String[] args) throws IOException {
        Socket socket = new Socket("127.0.0.1", 1145);
        System.out.println("建立连接：" + socket);
    }
}

// Invariants: mark <= position <= limit <= capacity
    private int mark = -1;
    private int position = 0;
    private int limit;
    private int capacity;

public Buffer flip() {
        limit = position;
        position = 0;
        mark = -1;
        return this;
    }

public class NioServer {
    public static void main(String[] args) {
        try {
            //1、打开ServerSocketChannel,用于监听客户端的连接，它是所有客户端连接的父管道(代表客户端连接的管道都是通过它创建的)
            ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
            //2、绑定监听端口，设置连接为非阻塞模式
            serverSocketChannel.socket().bind(new InetSocketAddress(1145));
            serverSocketChannel.configureBlocking(false);
            //3、创建多路复用器Selector
            Selector selector = Selector.open();
            //4、将ServlerSocketChannel注册到selector上，监听客户端连接事件ACCEPT
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
            //5、创建 Reactor线程，让多路复用器在 Reactor 线程中执行多路复用程序
            new Thread(new SingleReactor(selector)).start();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

class SingleReactor implements  Runnable{

    private final Selector selector;
    public SingleReactor(Selector selector) {
        this.selector = selector;
    }

    public void run() {
        //6、selector轮询准备就绪的事件
        while (true) {
            try {
                selector.select(1000);
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                while (iterator.hasNext()) {
                    SelectionKey selectionKey = iterator.next();
                    iterator.remove();
                    try {
                        processKey(selectionKey);
                    } catch (IOException e) {
                        e.printStackTrace();
                        if (selectionKey !=null ) {
                            selectionKey.cancel();
                            SelectableChannel channel = selectionKey.channel();
                            if (channel !=null) {
                                channel.close();
                            }
                        }
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    private void processKey(SelectionKey key) throws IOException {
        if (key.isValid()) {
            //7、根据准备就绪的事件类型分别处理
            if (key.isAcceptable()) {//客户端请求连接事件就绪
                //7.1、接收一个新的客户端连接，创建对应的SocketChannel,
                ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
                SocketChannel socketChannel = serverSocketChannel.accept();
                //7.2、设置socketChannel的非阻塞模式，并将其注册到Selector上，监听读事件
                socketChannel.configureBlocking(false);
                socketChannel.register(this.selector,SelectionKey.OP_READ);
            }
            if (key.isReadable()) {//读事件准备继续
                //7.1、读客户端发送过来的数据
                SocketChannel socketChannel = (SocketChannel) key.channel();
                ByteBuffer readBufer = ByteBuffer.allocate(1024);
                int readBytes = socketChannel.read(readBufer);
                //前面设置过socketChannel是非阻塞的，故要通过返回值判断读取到的字节数
                if (readBytes > 0) {
                    readBufer.flip();//读写模式切换
                    byte[] bytes = new byte[readBufer.remaining()];
                    readBufer.get(bytes);
                    String msg = new String(bytes,"utf-8");
                    //进行业务处理
                    String response = doService(msg);
                    //给客户端响应数据
                    System.out.println("服务端开始向客户端响应数据");
                    byte[] responseBytes = response.getBytes();
                    ByteBuffer writeBuffer = ByteBuffer.allocate(responseBytes.length);
                    writeBuffer.put(responseBytes);
                    writeBuffer.flip();
                    socketChannel.write(writeBuffer);
                }else if (readBytes < 0) {
                    //值为-1表示链路通道已经关闭
                    key.cancel();
                    socketChannel.close();
                }else {
                    //没读取到数据，忽略
                }
            }
        }
    }

    private String doService(String msg) {
        System.out.println("成功接收来自客户端发送过来的数据:"+msg);
        return "hello client,i am server!";
    }

}

public class NioClient {

    public static void main(String[] args) {
        try {
            //1、窗口客户端SocketChannel,绑定客户端本地地址(不选默认随机分配一个可用地址)
            SocketChannel socketChannel = SocketChannel.open();
            //2、设置非阻塞模式,
            socketChannel.configureBlocking(false);
            //3、创建Selector
            Selector selector = Selector.open();
            //3、创建Reactor线程
            new Thread(new SingleReactorClient(socketChannel,selector)).start();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

class SingleReactorClient implements  Runnable{
    private final SocketChannel socketChannel;
    private final Selector selector;

    public SingleReactorClient(SocketChannel socketChannel, Selector selector) {
        this.socketChannel = socketChannel;
        this.selector = selector;
    }

    public void run() {
        try {
            //连接服务端
            doConnect(socketChannel,selector);
        } catch (IOException e) {
            e.printStackTrace();
            System.exit(1);
        }

        //5、多路复用器执行多路复用程序
        while (true) {
            try {
                selector.select(1000);
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                while (iterator.hasNext()) {
                    SelectionKey selectionKey = iterator.next();
                    processKey(selectionKey);
                    iterator.remove();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    private void doConnect(SocketChannel sc, Selector selector) throws IOException {
        System.out.println("客户端成功启动,开始连接服务端");
        //3、连接服务端
        boolean connect = sc.connect(new InetSocketAddress("127.0.0.1", 1145));
        //4、将socketChannel注册到selector并判断是否连接成功，连接成功监听读事件，没有继续监听连接事件
        System.out.println("connect="+connect);
        if (connect) {
            sc.register(selector, SelectionKey.OP_READ);
            System.out.println("客户端成功连上服务端,准备发送数据");
            //开始进行业务处理，向服务端发送数据
            doService(sc);
        }else {
            sc.register(selector,SelectionKey.OP_CONNECT);
        }
    }

    private void processKey(SelectionKey key) throws IOException {
        if (key.isValid()) {
            //6、根据准备就绪的事件类型分别处理
            if (key.isConnectable()) {//服务端可连接事件准备就绪
                SocketChannel sc = (SocketChannel) key.channel();
                if (sc.finishConnect()) {
                    //6.1、向selector注册可读事件(接收来自服务端的数据)
                    sc.register(selector,SelectionKey.OP_READ);
                    //6.2、处理业务 向服务端发送数据
                    doService(sc);
                }else {
                    //连接失败，退出
                    System.exit(1);
                }
            }

            if (key.isReadable()) {//读事件准备继续
                //6.1、读服务端返回的数据
                SocketChannel sc = (SocketChannel) key.channel();
                ByteBuffer readBufer = ByteBuffer.allocate(1024);
                int readBytes = sc.read(readBufer);
                //前面设置过socketChannel是非阻塞的，故要通过返回值判断读取到的字节数
                if (readBytes > 0) {
                    readBufer.flip();//读写模式切换
                    byte[] bytes = new byte[readBufer.remaining()];
                    readBufer.get(bytes);
                    String msg = new String(bytes,"utf-8");
                    //接收到服务端返回的数据后进行相关操作
                    doService(msg);
                }else if (readBytes < 0) {
                    //值为-1表示链路通道已经关闭
                    key.cancel();
                    sc.close();
                }else {
                    //没读取到数据，忽略
                }
            }
        }
    }
    private static void doService(SocketChannel socketChannel) throws IOException {
        System.out.println("客户端开始向服务端发送数据:");
        //向服务端发送数据
        byte[] bytes = "hello nioServer,i am nioClient !".getBytes();
        ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
        writeBuffer.put(bytes);
        writeBuffer.flip();
        socketChannel.write(writeBuffer);
    }

    private String doService(String msg) {
        System.out.println("成功接收来自服务端响应的数据:"+msg);
        return "";
    }
}

<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.42.Final</version>
</dependency>

<dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-simple</artifactId>
    <version>1.7.25</version>
</dependency>
<dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-api</artifactId>
    <version>1.7.25</version>
</dependency>

public class NettyServer {

    private static final Logger log = LoggerFactory.getLogger(NettyServer.class);

    public static void main(String[] args) {
        NettyServer NettyServer = new NettyServer();
        NettyServer.start(8888);
    }

    public void start(int port) {
        //创建 bossGroup workerGroup 分别管理连接建立事件和具体的业务处理事件
        EventLoopGroup boss = new NioEventLoopGroup();
        EventLoopGroup worker = new NioEventLoopGroup();
        try {
            //创建启动引导类
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            //配置参数
            serverBootstrap.group(boss,worker)
                    .channel(NioServerSocketChannel.class)       //指定服务端通道，用于接收并创建新连接
                    .handler(new LoggingHandler(LogLevel.DEBUG)) // 给 boss group 配置 handler
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        //每个客户端 channel 初始化时都会执行该方法来配置该 channel 的相关 handler
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            //获取与该 channel 绑定的 pipeline
                            ChannelPipeline pipeline = ch.pipeline();
                            //向 pipeline 中添加 handler，如果没有注册到这里则不会生效
                            pipeline.addLast(new ServerOutboundHandler1());
                            pipeline.addLast(new ServerInboundHandler1());
                            pipeline.addLast(new ServerInboundHandler2());
                        }
                    }); //给 worker group 配置 handler
            //服务端绑定端口启动
            ChannelFuture future = serverBootstrap.bind(port).sync();
            //服务端监听端口关闭
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            log.error("netty server error ,{}",e.getMessage());
        } finally {
            //优雅关闭 boss worker
            boss.shutdownGracefully();
            worker.shutdownGracefully();
        }
    }
}

public class ServerInboundHandler1 extends ChannelInboundHandlerAdapter {
    private static final Logger log = LoggerFactory.getLogger(ServerInboundHandler1.class);

    /**
     * 通道准备就绪时
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("ServerInboundHandler1 channelActive-----");

        //将事件向下传递
        ctx.fireChannelActive();
    }

    /**
     * 通道有数据可读时
     *
     * @param ctx
     * @param msg
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        log.info("ServerInboundHandler1 channelRead----,remoteAddress={}", ctx.channel().remoteAddress());
        //处理接收的数据
        ByteBuf buf = (ByteBuf) msg;
        log.info("ServerInboundHandler1:received client data = {}", buf.toString(StandardCharsets.UTF_8));

        //将事件消息向下传递，如果不传递则 msg 不会到达下一个 handler
        ctx.fireChannelRead(msg);
    }

    /**
     * 数据读取完毕时
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        log.info("channelReadComplete----");
        //数据读取结束后向客户端写回数据
        byte[] data = "hello client , i am server".getBytes(StandardCharsets.UTF_8);
        ByteBuf buffer = Unpooled.buffer(data.length);
        buffer.writeBytes(data);//以bytebuf为中心,看是写到bytebuf中还是从bytebuf中读
        ByteBuf buf = Unpooled.copiedBuffer("hello client , i am server", StandardCharsets.UTF_8);
        ctx.writeAndFlush(buf);//通过ctx写，事件会从当前handler向pipeline头部移动
        //ctx.channel().writeAndFlush(buf);//通过Channel写,事件会从通道尾部向头部移动
    }

    /**
     * 发生异常时
     *
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.info("ServerInboundHandler1 exceptionCaught----,cause={}", cause.getMessage());
    }
}

public class ServerInboundHandler2 extends ChannelInboundHandlerAdapter {
    private static final Logger log = LoggerFactory.getLogger(ServerInboundHandler2.class);

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("ServerInboundHandler2 channelActive-----");
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        log.info("ServerInboundHandler2 channelRead----,remoteAddress={}", ctx.channel().remoteAddress());
        //处理接收的数据
        ByteBuf buf = (ByteBuf) msg;
        log.info("ServerInboundHandler2:received client data = {}", buf.toString(StandardCharsets.UTF_8));
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        log.info("ServerInboundHandler2 channelReadComplete----");
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    }
}

public class ServerInboundHandler1 extends ChannelInboundHandlerAdapter {
    private static final Logger log = LoggerFactory.getLogger(ServerInboundHandler1.class);

    /**
     * 通道准备就绪时
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("ServerInboundHandler1 channelActive-----");

        //将事件向下传递
        //ctx.fireChannelActive();
        super.channelActive(ctx);
    }

    /**
     * 通道有数据可读时
     *
     * @param ctx
     * @param msg
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        log.info("ServerInboundHandler1 channelRead----,remoteAddress={}", ctx.channel().remoteAddress());
        //处理接收的数据
        ByteBuf buf = (ByteBuf) msg;
        log.info("ServerInboundHandler1:received client data = {}", buf.toString(StandardCharsets.UTF_8));

        //将事件消息向下传递，如果不传递则 msg 不会到达下一个 handler
        ctx.fireChannelRead(msg);
//        super.channelRead(ctx, msg);
    }

    /**
     * 数据读取完毕时
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        log.info("channelReadComplete----");
        //数据读取结束后向客户端写回数据
        byte[] data = "hello client , i am server".getBytes(StandardCharsets.UTF_8);
        ByteBuf buffer = Unpooled.buffer(data.length);
        buffer.writeBytes(data);//以bytebuf为中心,看是写到bytebuf中还是从bytebuf中读
        ByteBuf buf = Unpooled.copiedBuffer("hello client , i am server", StandardCharsets.UTF_8);
        ctx.writeAndFlush(buf);//通过ctx写，事件会从当前handler向pipeline头部移动
        //ctx.channel().writeAndFlush(buf);//通过Channel写,事件会从通道尾部向头部移动
    }

    /**
     * 发生异常时
     *
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.info("ServerInboundHandler1 exceptionCaught----,cause={}", cause.getMessage());
    }
}

public class NettyClient {

    private static final Logger log = LoggerFactory.getLogger(NettyClient.class);

    public static void main(String[] args) {
        NettyClient client = new NettyClient();
        client.start("127.0.0.1", 8888);
    }

    public void start(String host, int port) {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            //添加客户端 channel 对应的 handler
                            pipeline.addLast(new ClientInboundHandler1());
                            pipeline.addLast(new ClientSimpleInboundHandler2());
                        }
                    });
            //连接远程启动
            ChannelFuture future = bootstrap.connect(host, port).sync();
            //监听通道关闭
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            log.error("netty client error ,msg={}", e.getMessage());
        } finally {
            //优雅关闭
            group.shutdownGracefully();
        }
    }
}

public class ClientInboundHandler1 extends ChannelInboundHandlerAdapter {
    private static final Logger log = LoggerFactory.getLogger(ClientInboundHandler1.class);

    /**
     * 通道准备就绪
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("ClientInboundHandler1 channelActive begin send data");
        //通道准备就绪后开始向服务端发送数据
        ByteBuf buf = Unpooled.copiedBuffer("hello server,i am client".getBytes(StandardCharsets.UTF_8));
        ctx.writeAndFlush(buf);
    }

    /**
     * 通道有数据可读（服务端返回了数据）
     *
     * @param ctx
     * @param msg
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        log.info("ClientInboundHandler1 channelRead");
        ByteBuf buf = (ByteBuf) msg;
        log.info("ClientInboundHandler1: received server data ={}", buf.toString(StandardCharsets.UTF_8));

        // 接着传递消息给下一个 ChannelInboundHandler
        ctx.fireChannelRead(msg);
    }

    /**
     * 数据读取完毕
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        super.channelReadComplete(ctx);
    }

    /**
     * 产生了异常
     *
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
    }
}

public class ClientSimpleInboundHandler2 extends SimpleChannelInboundHandler<ByteBuf> {
    private static final Logger log = LoggerFactory.getLogger(ClientSimpleInboundHandler2.class);

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
        log.info("ClientSimpleInboundHandler2 channelRead");
        log.info("ClientSimpleInboundHandler2: received server data ={}", msg.toString(StandardCharsets.UTF_8));
    }

}

// 主线程，不处理任何业务逻辑，只是接收客户的连接请求
EventLoopGroup boss = new NioEventLoopGroup();
// 工作线程，处理注册其上 Channel的 I/O 事件及其他 Task
EventLoopGroup worker = new NioEventLoopGroup();

private static final int DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt("io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));

 .childHandler(new ChannelInitializer<SocketChannel>() {
                        //每个客户端 channel 初始化时都会执行该方法来配置该 channel 的相关 handler
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            //获取与该 channel 绑定的 pipeline
                            ChannelPipeline pipeline = ch.pipeline();
                            //向 pipeline 中添加 handler，如果没有注册到这里则不会生效
                            pipeline.addLast(new ServerOutboundHandler1());
                            pipeline.addLast(new ServerInboundHandler1());
                            pipeline.addLast(new ServerInboundHandler2());
                        }

public class NettyServer {

    private static final Logger log = LoggerFactory.getLogger(NettyServer.class);

    public static void main(String[] args) {
        NettyServer NettyServer = new NettyServer();
        NettyServer.start(8888);
    }

    public void start(int port) {
        //创建 bossGroup workerGroup 分别管理连接建立事件和具体的业务处理事件
        EventLoopGroup boss = new NioEventLoopGroup();
        EventLoopGroup worker = new NioEventLoopGroup();

        // 只创建一次 serverInboundHandler2 对象
        ServerInboundHandler2 serverInboundHandler2 = new ServerInboundHandler2();

        try {
            //创建启动引导类
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            //配置参数
            serverBootstrap.group(boss,worker)
                    .channel(NioServerSocketChannel.class)       //指定服务端通道，用于接收并创建新连接
                    .handler(new LoggingHandler(LogLevel.DEBUG)) // 给 boss group 配置 handler
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        //每个客户端 channel 初始化时都会执行该方法来配置该 channel 的相关 handler
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            //获取与该 channel 绑定的 pipeline
                            ChannelPipeline pipeline = ch.pipeline();
                            //向 pipeline 中添加 handler，如果没有注册到这里则不会生效
                            pipeline.addLast(new ServerOutboundHandler1());
                            pipeline.addLast(new ServerInboundHandler1());

                            // 在这里对 serverInboundHandler2 进行复用
                            pipeline.addLast(serverInboundHandler2);

                        }
                    }); //给 worker group 配置 handler
            ChannelFuture future = serverBootstrap.bind(port).sync();
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            log.error("netty server error ,{}",e.getMessage());
        } finally {
            //优雅关闭 boss worker
            boss.shutdownGracefully();
            worker.shutdownGracefully();
        }
    }
}

[nioEventLoopGroup-2-1] INFO handler.client.ClientInboundHandler1 - ClientInboundHandler1 channelActive begin send data
[nioEventLoopGroup-2-1] WARN io.netty.channel.DefaultChannelPipeline - An exceptionCaught() event was fired, and it reached at the tail of the pipeline. It usually means the last handler in the pipeline did not handle the exception.
java.io.IOException: 你的主机中的软件中止了一个已建立的连接。
    at java.base/sun.nio.ch.SocketDispatcher.read0(Native Method)
    at java.base/sun.nio.ch.SocketDispatcher.read(SocketDispatcher.java:46)
    at java.base/sun.nio.ch.IOUtil.readIntoNativeBuffer(IOUtil.java:330)
    at java.base/sun.nio.ch.IOUtil.read(IOUtil.java:284)
    at java.base/sun.nio.ch.IOUtil.read(IOUtil.java:259)
    at java.base/sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:417)
    at io.netty.buffer.PooledByteBuf.setBytes(PooledByteBuf.java:247)
    at io.netty.buffer.AbstractByteBuf.writeBytes(AbstractByteBuf.java:1147)
    at io.netty.channel.socket.nio.NioSocketChannel.doReadBytes(NioSocketChannel.java:347)
    at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:148)
    at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:700)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:635)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:552)
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:514)
    at io.netty.util.concurrent.SingleThreadEventExecutor$6.run(SingleThreadEventExecutor.java:1044)
    at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
    at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
    at java.base/java.lang.Thread.run(Thread.java:840)

Process finished with exit code 0

@ChannelHandler.Sharable
public class ServerInboundHandler2 extends ChannelInboundHandlerAdapter {
    private static final Logger log = LoggerFactory.getLogger(ServerInboundHandler2.class);

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("ServerInboundHandler2 channelActive-----");
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        log.info("ServerInboundHandler2 channelRead----,remoteAddress={}", ctx.channel().remoteAddress());
        //处理接收的数据
        ByteBuf buf = (ByteBuf) msg;
        log.info("ServerInboundHandler2:received client data = {}", buf.toString(StandardCharsets.UTF_8));
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        log.info("ServerInboundHandler2 channelReadComplete----");
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    }
}

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        log.info("ServerInboundHandler1 channelRead----,remoteAddress={}", ctx.channel().remoteAddress());
        //处理接收的数据
        ByteBuf buf = (ByteBuf) msg;
        log.info("ServerInboundHandler1:received client data = {}", buf.toString(StandardCharsets.UTF_8));

        //将事件消息向下传递，如果不传递则 msg 不会到达下一个 handler
        ctx.fireChannelRead(msg);
    }

    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        boolean release = true;
        try {
            // 判断是否是接收
            if (this.acceptInboundMessage(msg)) {
                // 调用 channelRead0 方法
                this.channelRead0(ctx, msg);
            } else {
                release = false;
                ctx.fireChannelRead(msg);
            }
        } finally {
            if (this.autoRelease && release) {
                // 对原始资源释放
                ReferenceCountUtil.release(msg);
            }

        }

    }

protected abstract void channelRead0(ChannelHandlerContext var1, I var2) throws Exception;

// 设置 HeapByteBuf 模式，但 ByteBuf 的分配器 ByteBufAllocator 要设置为非池化，否则不能切换到堆缓冲器模式
System.setProperty("io.netty.noUnsafe","true");

//引导类中设置非池化模式
bootstrap.childOption(ChannelOption.ALLOCATOR,UnpooledByteBufAllocator.DEFAULT)
//或者通过系统参数设置
System.setProperty("io.netty.allocator.type","pooled");
System.setProperty("io.netty.allocator.type","unpooled");

public interface Future<V> extends java.util.concurrent.Future<V> {
  // 只有IO操作完成时才返回true
  boolean isSuccess();
  // 只有当cancel(boolean)成功取消时才返回true
  boolean isCancellable();
  // IO操作发生异常时，返回导致IO操作以此的原因，如果没有异常，返回null
  Throwable cause();
  // 向Future添加事件，future完成时，会执行这些事件，如果add时future已经完成，会立即执行监听事件
  Future<V> addListener(GenericFutureListener<? extends Future<? super V>> listener);
  Future<V> addListeners(GenericFutureListener<? extends Future<? super V>>... listeners);
  // 移除监听事件，future完成时，不会触发
  Future<V> removeListener(GenericFutureListener<? extends Future<? super V>> listener);
  Future<V> removeListeners(GenericFutureListener<? extends Future<? super V>>... listeners);
  // 等待future done
  Future<V> sync() throws InterruptedException;
  // 等待future done，不可打断
  Future<V> syncUninterruptibly();
  // 等待future完成
  Future<V> await() throws InterruptedException;
  // 等待future 完成，不可打断
  Future<V> awaitUninterruptibly();
  boolean await(long timeout, TimeUnit unit) throws InterruptedException;
  boolean await(long timeoutMillis) throws InterruptedException;
  boolean awaitUninterruptibly(long timeout, TimeUnit unit);
  boolean awaitUninterruptibly(long timeoutMillis);
  // 立刻获得结果，如果没有完成，返回null
  V getNow();
  // 如果成功取消，future会失败，导致CancellationException
  @Override
  boolean cancel(boolean mayInterruptIfRunning);

package netty;

import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.Promise;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.LocalDateTime;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

public class NettyFutureTest {
    private static final Logger log = LoggerFactory.getLogger(NettyFutureTest.class);

    @Test
    public void testFuture() throws InterruptedException, ExecutionException {
        EventLoopGroup group = new NioEventLoopGroup();

        Future<String> future = group.submit( () ->{
            log.info("---异步线程执行任务开始----,time={}", LocalDateTime.now().toString());
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            log.info("---异步线程执行任务结束----,time={}", LocalDateTime.now().toString());
            return "hello netty future";
        });
        /*String result = future.get();
        log.info("----主线程阻塞等待异步线程执行结果:{}",result);*/
        //设置监听
        future.addListener( future1 -> {
            log.info("---收到异步线程执行任务结果通知----执行结果是;{},time={}",future1.get(), LocalDateTime.now().toString());
        });
        log.info("---主线程----");
        TimeUnit.SECONDS.sleep(10);
    }

    @Test
    public void testPromise() throws InterruptedException {
        EventLoopGroup group = new NioEventLoopGroup();

        Promise promise = new DefaultPromise(group.next());//promise绑定到eventloop上

        group.submit(()->{
            log.info("---异步线程执行任务开始----,time={}", LocalDateTime.now().toString());
            try {
                int i = 1/0;
                TimeUnit.SECONDS.sleep(3);
                promise.setSuccess("hello netty promise");
                TimeUnit.SECONDS.sleep(3);
                log.info("---异步线程执行任务结束----,time={}", LocalDateTime.now().toString());
                return;
            } catch (Exception e) {
                promise.setFailure(e);
            }
        });
        //设置监听回调
        promise.addListener(future -> {
            log.info("----异步任务执行结果:{}",future.isSuccess());
        });
        promise.addListener(future2 -> {
            log.info("----异步任务执行结果:{}",future2.isSuccess());
        });
        log.info("---主线程----");
        TimeUnit.SECONDS.sleep(10);
    }

}

public class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();
        pipeline.addLast(new FixedLengthFrameDecoder(10));
        pipeline.addLast("decoder", new StringDecoder());
        pipeline.addLast("encoder", new StringEncoder());
        // 自己的逻辑Handler
        pipeline.addLast("handler", new PeServerHandler());
    }
}

public class ClientChannelInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();
        pipeline.addLast(new FixedLengthFrameDecoder(10));
        pipeline.addLast("decoder", new StringDecoder());
        pipeline.addLast("encoder", new StringEncoder());
        // 客户端的逻辑
        pipeline.addLast("handler", new PeClientHandler());
    }
}

public class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();
        pipeline.addLast(new LineBasedFrameDecoder(2048));
        pipeline.addLast("decoder", new StringDecoder());
        pipeline.addLast("encoder", new StringEncoder());
        // 自己的逻辑Handler
        pipeline.addLast("handler", new PeServerHandler());
    }
}

public class ClientChannelInitializer extends  ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();
        pipeline.addLast(new LineBasedFrameDecoder(2048));
        pipeline.addLast("decoder", new StringDecoder());
        pipeline.addLast("encoder", new StringEncoder());
        // 客户端的逻辑
        pipeline.addLast("handler", new PeClientHandler());
    }
}

public class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();
        pipeline.addLast(new LengthFieldBasedFrameDecoder(1024, // 帧的最大长度，即每个数据包最大限度
                0, // 长度字段偏移量
                4, // 长度字段所占的字节数
                0, // 消息头的长度，可以为负数
                4) // 需要忽略的字节数，从消息头开始，这里是指整个包

        );
//        pipeline.addLast("decoder", new StringDecoder());
        //pipeline.addLast("encoder", new StringEncoder());
        // 自己的逻辑Handler
        pipeline.addLast("handler", new PeServerHandler());
    }
}

public class ClientChannelInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();
        pipeline.addLast(new LineBasedFrameDecoder(10));
//        pipeline.addLast("decoder", new StringDecoder());
//        pipeline.addLast("encoder", new StringEncoder());
        // 客户端的逻辑
        pipeline.addLast("handler", new PeClientHandler());
    }
}

public class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();
        pipeline.addLast(new MyProtocolEncoder());

        /**
         *
         * MsgReq 对象 ：
         * byte type 字段占一个字节
         * int length 字段 占4个字节
         * 剩下的是消息体
         * 那么 这里的 参数设置：
         *
         * int maxFrameLength：可以设置你认为的小题最大上限
         * int lengthFieldOffset：长度字段从哪个位置开始读：第0个字节是type，长度是从1开始的
         * int lengthFieldLength：长度字段所占的字节数，int类型占4个字节
         * int lengthAdjustment：是否需要调整消息头的长度，即读取消息头是否需要偏移一下，我们这里不需要
         * int initialBytesToStrip：消息体是否需要忽略一些字节数，比如忽略消息头的长度，我们这里消息头也算在对象里面了所以不忽略
         * boolean failFast：如果读取长度不够是否快速失败
         *
         */
        pipeline.addLast(new MyProtocolDecoder(1024, // 帧的最大长度，即每个数据包最大限度
                1, // 长度字段偏移量
                4, // 长度字段所占的字节数
                0, // 消息头的长度，可以为负数
                0, // 需要忽略的字节数，从消息头开始，这里是指整个包
                true)

        );
        // 自己的逻辑Handler
        pipeline.addLast("handler", new PeServerHandler());
    }
}

public class ClientChannelInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();
        pipeline.addLast(new MyProtocolEncoder());
        /**
         *
         * MsgReq 对象 ：
         * byte type 字段占一个字节
         * int length 字段 占4个字节
         * 剩下的是消息体
         * 那么 这里的 参数设置：
         *
         * int maxFrameLength：可以设置你认为的小题最大上限
         * int lengthFieldOffset：长度字段从哪个位置开始读：第0个字节是type，长度是从1开始的
         * int lengthFieldLength：长度字段所占的字节数，int类型占4个字节
         * int lengthAdjustment：是否需要调整消息头的长度，即读取消息头是否需要偏移一下，我们这里不需要
         * int initialBytesToStrip：消息体是否需要忽略一些字节数，比如忽略消息头的长度，我们这里消息头也算在对象里面了所以不忽略
         * boolean failFast：如果读取长度不够是否快速失败
         *
         */
        pipeline.addLast(new MyProtocolDecoder(1024,
                1,
                4,
                0,
                0,
                true));
        // 客户端的逻辑
        pipeline.addLast("handler", new PeClientHandler());
    }
}

protected void initChannel(SocketChannel ch) throws Exception {
    //获取与该channel绑定的pipeline
    ChannelPipeline pipeline = ch.pipeline();

    //测试 http service
    pipeline.addLast(new HttpResponseEncoder());

    pipeline.addLast(new HttpRequestDecoder());
    //文件上传需要设置大点儿 单位是字节
    pipeline.addLast(new HttpObjectAggregator(1024*1024*8));
    pipeline.addLast(new MyHttpServerHandler());
}

public class MyHttpServerHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
    private static final Logger log = LoggerFactory.getLogger(MyHttpServerHandler.class);

    private static final HttpDataFactory HTTP_DATA_FACTORY = new DefaultHttpDataFactory(DefaultHttpDataFactory.MAXSIZE);

    static {
        DiskFileUpload.baseDirectory = "/opt/netty/fileupload";
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest fullHttpRequest) throws Exception {
        //HttpRequest request = fullHttpRequest;
        //String uri = fullHttpRequest.uri(); 
        //获取method
        HttpMethod method = fullHttpRequest.method();
        //根据method解析参数，封装数据，
        if (HttpMethod.GET.equals(method)) {
           parseGet(fullHttpRequest);
        }else if (HttpMethod.POST.equals(method)) {
            parsePost(fullHttpRequest);
        }else {
            log.error("{} method is not supported ,please change http method for get or post!");
        }
        //service
        //response client
        StringBuilder sb = new StringBuilder();
        sb.append("<html>");
        sb.append("<head>");
        sb.append("</head>");
        sb.append("<body>");
        sb.append("<h3>success</h3>");
        sb.append("</body>");
        sb.append("</html>");
        writeResponse(ctx,fullHttpRequest,HttpResponseStatus.OK,sb.toString());
    }

    private void writeResponse(ChannelHandlerContext ctx, FullHttpRequest fullHttpRequest, HttpResponseStatus status, String msg) {
        FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,status);
        response.content().writeBytes(msg.getBytes(StandardCharsets.UTF_8));
        response.headers().set(HttpHeaderNames.CONTENT_TYPE,"text/html;charset=utf-8");
        HttpUtil.setContentLength(response,response.content().readableBytes());
        boolean keepAlive = HttpUtil.isKeepAlive(fullHttpRequest);
        if (keepAlive) {
            response.headers().set(HttpHeaderNames.CONNECTION,"keep-alive");
            ctx.writeAndFlush(response);
        }else {
            ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
        }
    }

    private void parsePost(FullHttpRequest fullHttpRequest) {
        //获取content-type
        String contentType = getContentType(fullHttpRequest);
        switch (contentType) {
            case "application/json":
                parseJson(fullHttpRequest.content());
                break;
            case "application/x-www-form-urlencoded":
                parseForm(fullHttpRequest);
                break;
            case "multipart/form-data":
                parseMultipart(fullHttpRequest);
                break;
            default:
        }

    }

    private void parseMultipart(FullHttpRequest fullHttpRequest) {
        HttpPostRequestDecoder postRequestDecoder = new HttpPostRequestDecoder(HTTP_DATA_FACTORY,fullHttpRequest);
        //判断是否是multipart
        if (postRequestDecoder.isMultipart()) {
        //获取 body中的数据
        List<InterfaceHttpData> bodyHttpDatas = postRequestDecoder.getBodyHttpDatas();
        bodyHttpDatas.forEach(dataItem ->{
            //获取数据项的类型
            InterfaceHttpData.HttpDataType dataType = dataItem.getHttpDataType();
            //判断是普通表达项还是文件上传项
            if (dataType.equals(InterfaceHttpData.HttpDataType.Attribute)) {
                //普通表单项 直接获取数据
                Attribute attribute = (Attribute) dataItem;
                try {
                    log.info("表单项名称:{},表单项值:{}",attribute.getName(),attribute.getValue());
                } catch (IOException e) {
                    log.error("获取表单项数据错误,msg={}",e.getMessage());
                }

            }else if (dataType.equals(InterfaceHttpData.HttpDataType.FileUpload)) {
                //文件上传项 处理待上传的数据
                FileUpload fileUpload = (FileUpload) dataItem;
                //获取原始文件名称
                String filename = fileUpload.getFilename();
                //获取表单name属性
                String name = fileUpload.getName();
                log.info("文件名称:{},表单项名称:{}",filename,name);
                //将文件数据保存到磁盘
                if (fileUpload.isCompleted()) {
                    try {
                        String path = DiskFileUpload.baseDirectory + File.separator + filename;
                        //File file = fileUpload.getFile();
                        fileUpload.renameTo(new File(path));
                    } catch (IOException e) {
                        log.error("文件转存失败,msg={}",e.getMessage());
                    }
                }
            }else {

            }
        });
        }
    }

    private void parseForm(FullHttpRequest fullHttpRequest) {
        //post请求时uri中也可能携带参数
        parseKVstr(fullHttpRequest.uri(),true);
        //解析请求体中的表单数据
        parseFormData(fullHttpRequest.content());
    }

    private void parseFormData(ByteBuf body) {
        String bodystr = body.toString(StandardCharsets.UTF_8);
        parseKVstr(bodystr,false);
    }

    private void parseJson(ByteBuf jsonbody) {
        String jsonstr = jsonbody.toString(StandardCharsets.UTF_8);
        //使用json工具反序列化
        JSONObject jsonObject = JSONObject.parseObject(jsonstr);
        //打印 json数据
        jsonObject.entrySet().stream().forEach(entry ->{
            log.info("json key={},json value={}",entry.getKey(),entry.getValue());
        });
    }

    private String getContentType(FullHttpRequest request) {
        HttpHeaders headers = request.headers();
        String contentType = headers.get(HttpHeaderNames.CONTENT_TYPE);// text/plain;charset=UTF-8
        //List<String> acceptEncoding = headers.getAll(HttpHeaderNames.ACCEPT_ENCODING);//accept-encoding:gzip, deflate, br
        return contentType.split(";")[0];
    }

    private void parseGet(FullHttpRequest request) {
        //通过uri解析请求参数
        parseKVstr(request.uri(),true);
    }

    private void parseKVstr(String str,boolean hasPath) {
        //通过QueryStringDecoder解析kv字符串
        QueryStringDecoder qsd = new QueryStringDecoder(str, StandardCharsets.UTF_8,hasPath);////get请求的uri是: path?k=v
        Map<String, List<String>> parameters = qsd.parameters();
        //封装参数，执行业务 此处打印即可
        if (parameters!=null && parameters.size() > 0) {
            parameters.entrySet().stream().forEach(entry->{
                log.info("参数名:{},参数值:{}",entry.getKey(),entry.getValue());
            });
        }
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        //ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("MyHttpServerHandler Exception,{}",cause.getMessage());
    }
}

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>测试 netty http server</title>
</head>
<body>
    <form action="http://localhost:8888/test" method="post" enctype="multipart/form-data">
        <label for="username">用户名:</label><input id="username" type="text" name="username"><br/>
        <label for="password">密码:</label><input id="password" type="password" name="password"><br/>
        <label for="email">邮箱:</label><input id="email" type="email" name="email"><br/>
        <label for="address">地址:</label><input id="address" type="text" name="address"><br/>
        <label for="pic">选择文件:</label><input id="pic" type="file" name="pic"><br/>

        <input type="submit" value="提交">
    </form>
</body>
</html>

.childHandler(new ChannelInitializer<SocketChannel>() {
    //每个客户端channel初始化时都会执行该方法来配置该channel的相关handler
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        //获取与该channel绑定的pipeline
        ChannelPipeline pipeline = ch.pipeline();
        //基于netty开发websocket 服务端
        pipeline.addLast(new HttpServerCodec());//HttpServerCodec = HttpRequestDecoder + HttpResponseEncoder
        pipeline.addLast(new HttpObjectAggregator(1024*1024*8));
        //pipeline.addLast(new ChunkedWriteHandler());
        //添加业务处理器
        pipeline.addLast(new MyWebSocketServerHandler());

    }
});

public class MyWebSocketServerHandler extends SimpleChannelInboundHandler<Object> {

    private static final Logger log = LoggerFactory.getLogger(MyWebSocketServerHandler.class);

    private WebSocketServerHandshaker serverHandshaker;

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
        //判断是http握手请求还是websocket请求
        if (msg instanceof FullHttpRequest) {
            boolean handShaker = handleHttpRequest(ctx, (FullHttpRequest) msg);
            if (handShaker) {
                //握手成功后 服务端主动推送消息，每隔5s推送一次
                new Thread(()->{
                    while (true) {
                        try {
                            ctx.channel().writeAndFlush(new TextWebSocketFrame("你好,这是服务器主动推送回来的数据,当前时间为:"+new Date().toString()));
                            TimeUnit.SECONDS.sleep(5);
                        } catch (InterruptedException e) {
                            log.error("push msg exception ,{}",e.getMessage());
                        }
                    }
                }).start();
            }
        }else if (msg instanceof WebSocketFrame) {
            handleWebSocketFrame(ctx,(WebSocketFrame)msg);
        }
    }

    /**
     * 接收到的消息是已经解码的WebSocketFrame消息
     * @param ctx
     * @param frame
     */
    private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
        // 判断链路消息类型
        if (frame instanceof CloseWebSocketFrame) { // 关闭链路指令
            serverHandshaker.close(ctx.channel(),((CloseWebSocketFrame) frame).retain());
            return;
        }
        if (frame instanceof PingWebSocketFrame) { // 维持链路的ping 指令
            ctx.channel().writeAndFlush(new PongWebSocketFrame(frame.content().retain()));
            return;
        }
        if (frame instanceof TextWebSocketFrame) { // 普通文本消息
            TextWebSocketFrame textFrame = (TextWebSocketFrame) frame;
            String message = textFrame.text();
            log.info("receive text msg is {}",message);
            //构造返回
            ctx.channel().writeAndFlush(new TextWebSocketFrame("你好,欢迎使用netty websocket 服务,当前时间为:"+new Date().toString()));
            return;
        }
        if (frame instanceof BinaryWebSocketFrame) { //二进制消息
            log.info("frame is binarywebsocketframe");
        }

    }

    private boolean handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest fullHttpRequest) {
        //先判断是否解码成功，
        if (!fullHttpRequest.decoderResult().isSuccess()) {
            sendHttpResponse(ctx,fullHttpRequest,new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
            return false;
        }
        // 然后判断是否要建立websocket连接
        //构造握手工厂 创建握手处理类，并且构造握手响应给客户端
        WebSocketServerHandshakerFactory serverHandshakerFactory = new WebSocketServerHandshakerFactory("ws://localhost:8888/mywebsocket",null,false);
        if (!fullHttpRequest.headers().contains(HttpHeaderNames.UPGRADE,"websocket",true)) {
            WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
            return false;
        }
        serverHandshaker = serverHandshakerFactory.newHandshaker(fullHttpRequest);
        serverHandshaker.handshake(ctx.channel(),fullHttpRequest);
        return true;
    }

    private void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest request, FullHttpResponse response) {
        if (!response.status().equals(HttpResponseStatus.OK)) {
            ByteBuf byteBuf = Unpooled.wrappedBuffer("error".getBytes(StandardCharsets.UTF_8));
            response.content().writeBytes(byteBuf);
            byteBuf.release();
            HttpUtil.setContentLength(response,response.content().readableBytes());
        }
        ChannelFuture future = ctx.channel().writeAndFlush(response);
        if (!HttpUtil.isKeepAlive(request) || !response.status().equals(HttpResponseStatus.OK) ) {
            future.addListener(ChannelFutureListener.CLOSE);
        }

    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("server error,msg is {}",cause.getMessage());
        ctx.close();
    }
}

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>test netty websocket </title>
</head>
<body>
<br>
<script type="text/javascript">
    var socket;
    if(!window.WebSocket){
        window.WebSocket=window.MozWebSocket;
    }
    if(window.WebSocket){
        socket=new WebSocket("ws://localhost:8888/mywebSocket");
        socket.onmessage=function(event){
            var ta=document.getElementById('responseText');
            ta.value="";
            ta.value=event.data;
        };
        socket.onopen=function(event){
            var ta=document.getElementById('responseText');
            ta.value='打开WebSocket服务器正常，浏览器支持WebSocket！';
        };
        socket.onclose=function(event){
            var ta=document.getElementById('responseText');
            ta.value='';
            ta.value="WebSocket 关闭！";
        };
    }else{
        alert("抱歉，您的浏览器不支持WebSocket协议！");
    }
    function send(message){
        if(!window.WebSocket){
            return;
        }
        if(socket!=null){
            socket.send(message);
        }else{
            alert("WebSocket连接没有建立成功，请刷新页面！");
        }
        /* if(socket.readyState==WebSocket.open){
            socket.send(message);
        }else{
            alert("WebSocket连接没有建立成功！");
        } */
    }
</script>
<form onsubmit="return false;">
    <input type="text" name="message" value="Netty WebSocket实战"/>
    <br><br>
    <input type="button" value="发送WebSocket请求消息" onclick="send(this.form.message.value)"/>
    <hr color="blue"/>
    <h3>服务端返回的应答消息</h3>
    <textarea id="responseText" style="width:500px;height:300px;"></textarea>
</form>
</body>
</html>

//Server端
protected void initChannel(SocketChannel ch) throws Exception {
    //获取与该channel绑定的pipeline
    ChannelPipeline pipeline = ch.pipeline();
    pipeline.addLast(new LoggingHandler(LogLevel.INFO));
    pipeline.addLast(new LengthFieldPrepender(2));
    pipeline.addLast(new StringEncoder());
    pipeline.addLast(new LengthFieldBasedFrameDecoder(1024,0,2,0,2));
    pipeline.addLast(new StringDecoder());
}

//Client端
protected void initChannel(SocketChannel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();
    pipeline.addLast(new LoggingHandler(LogLevel.INFO));
    pipeline.addLast(new LengthFieldPrepender(2));
    pipeline.addLast(new StringEncoder());
    pipeline.addLast(new LengthFieldBasedFrameDecoder(1024,0,2,0,2));
    pipeline.addLast(new StringDecoder());
}

@Slf4j
public class ServerReadIdleCheckHandler extends IdleStateHandler {

    public ServerReadIdleCheckHandler() {
        super(10, 0, 0, TimeUnit.SECONDS);
    }

    @Override
    protected void channelIdle(ChannelHandlerContext ctx, IdleStateEvent evt) throws Exception {
        log.info("server channel idle----");
        if (evt == IdleStateEvent.FIRST_READER_IDLE_STATE_EVENT) {
            ctx.close();
            log.info("server read idle , close channel.....");
            return;
        }
        super.channelIdle(ctx, evt);
    }
}

protected void initChannel(SocketChannel ch) throws Exception {
    //获取与该channel绑定的pipeline
    ChannelPipeline pipeline = ch.pipeline();
    pipeline.addLast(new LoggingHandler(LogLevel.INFO));
    pipeline.addLast(new ServerReadIdleCheckHandler());//添加ServerReadIdleCheckHandler
    pipeline.addLast(new LengthFieldPrepender(2));
    pipeline.addLast(new StringEncoder());
    pipeline.addLast(new LengthFieldBasedFrameDecoder(1024,0,2,0,2));
    pipeline.addLast(new StringDecoder());
}

@Slf4j
public class ClientWriteCheckIdleHandler extends IdleStateHandler {
    public ClientWriteCheckIdleHandler() {
        super(0, 5, 0, TimeUnit.SECONDS);
    }

    //也可在channelIdle方法中直接处理
}

@Slf4j
@ChannelHandler.Sharable
public class KeepaliveHandler extends ChannelDuplexHandler {

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt == IdleStateEvent.FIRST_WRITER_IDLE_STATE_EVENT) {
            log.info("client write idle,so send keepalive msg to server");
            ctx.writeAndFlush("this is keepalive msg");
        }

        super.userEventTriggered(ctx, evt);
    }
}

protected void initChannel(SocketChannel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();
    pipeline.addLast(new LoggingHandler(LogLevel.INFO));
    pipeline.addLast(new ClientWriteCheckIdleHandler());//write idle监测
    pipeline.addLast(new LengthFieldPrepender(2));
    pipeline.addLast(new StringEncoder());

    pipeline.addLast(new KeepaliveHandler());//发送keepalvie消息
    pipeline.addLast(new LengthFieldBasedFrameDecoder(1024,0,2,0,2));
    pipeline.addLast(new StringDecoder());
}