public class EchoServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println(ctx.channel().remoteAddress() + " -> Server :" + msg);

        // 写消息到管道
        ctx.write(msg); //写消息
        ctx.flush(); // 冲刷消息

        // 上面两个方法等同于 ctx.writeAndFlush(msg);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        // 当出现异常就关闭连接
        cause.printStackTrace();
        ctx.close();
    }
}

public class EchoServer {

    public static int DEFAULT_PORT = 7;

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

        try {
            port = Integer.parseInt(args[0]);
        } catch (RuntimeException ex) {
            port = DEFAULT_PORT;
        }

        // 多线程时间循环器
        NioEventLoopGroup boosGroup = new NioEventLoopGroup();
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            // 启动NIO 服务的引导程序类
            ServerBootstrap b = new ServerBootstrap();

            // 设置 EventLoopGroup
            b.group(boosGroup, workerGroup)                                 // 设置EventLoopGroup
                    .channel(NioServerSocketChannel.class)                  // 指明新的Channel的类型
                    .childHandler(new EchoServerHandler())                  // 指定ChannelHandler
                    .option(ChannelOption.SO_BACKLOG, 128)           // 设置的ServerChannel的一些选项
                    .childOption(ChannelOption.SO_KEEPALIVE, true);  // 设置的ServerChannel的子Channel的选项

            // 绑定端口，开始接收进来的连接
            ChannelFuture f = b.bind(port).sync();
            System.out.println("EchoServer已启动，端口：" + port);

            // 等待服务器 socket 关闭 。
            // 在这个例子中，这不会发生，但你可以优雅地关闭你的服务器。
            f.channel().closeFuture().sync();
        } finally {

            // 优雅的关闭
            workerGroup.shutdownGracefully();
            boosGroup.shutdownGracefully();
        }
    }
}

public class EchoClientHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

        // 从管道读消息
        ByteBuf buf = (ByteBuf) msg;    // 转为ByteBuf类型
        String m = buf.toString(CharsetUtil.UTF_8);//转换为字符串

        System.out.println("echo: " + m);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        // 当出现异常就关闭连接
        cause.printStackTrace();
        ctx.close();
    }
}

public final class EchoClient {

    public static void main(String[] args) {
        if (args.length != 2) {
            System.err.println("用法: java EchoClient <host name> <port number>");
            System.exit(1);
        }

        String hostName = args[0];
        int portNumber = Integer.parseInt(args[1]);

        // 配置客户端
        EventLoopGroup group = new NioEventLoopGroup();

        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .handler(new EchoClientHandler());

            // 连接到服务器
            ChannelFuture future = bootstrap.connect(hostName, portNumber).sync();
            Channel channel = future.channel();
            ByteBuffer writeBuffer = ByteBuffer.allocate(32);
            try (BufferedReader stdIn = new BufferedReader(new InputStreamReader(System.in))) {
                String userInput;
                while ((userInput = stdIn.readLine()) != null) {
                    writeBuffer.put(userInput.getBytes());
                    writeBuffer.flip();
                    writeBuffer.rewind();

                    // 转为ByteBuf
                    ByteBuf buf = Unpooled.copiedBuffer(writeBuffer);
                    // 写消息到管道
                    channel.writeAndFlush(buf);
                    // 清理缓存
                    writeBuffer.clear();
                }
            }
        } catch (UnknownHostException e) {
            System.err.println("不明主机，主机名为： " + hostName);
            System.exit(1);
        } catch (IOException e) {
            System.err.println("不能从主机中获取I/O，主机名为：" + hostName);
            System.exit(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            // 优雅的关闭
            group.shutdownGracefully();
        }
    }
}

public class ByteBufferDemo {

    public static void main(String[] args) {
        // 创建一个缓冲区
        ByteBuffer buffer = ByteBuffer.allocate(10);
        System.out.println("------------初始时缓冲区------------");
        printBuffer(buffer);

        // 添加一些数据到缓冲区中
        System.out.println("------------添加数据到缓冲区------------");
        String s = "love";
        buffer.put(s.getBytes());
        printBuffer(buffer);

        // 切换成读模式
        System.out.println("------------执行flip切换到读取模式------------");
        buffer.flip();
        printBuffer(buffer);

        // 读取数据
        System.out.println("------------读取数据------------");
        // 创建一个limit()大小的字节数组(因为就只有limit这么多个数据可读)
        byte[] bytes = new byte[buffer.limit()];
        // 将读取的数据装进我们的字节数组中
        buffer.get(bytes);
        printBuffer(buffer);

        // 执行compact
        System.out.println("------------执行compact------------");
        buffer.compact();
        printBuffer(buffer);

        // 执行clear
        System.out.println("------------执行clear清空缓冲区------------");
        buffer.clear();
        printBuffer(buffer);
    }

    /**
     * 打印出ByteBuffer的信息
     *
     * @param buffer
     */
    private static void printBuffer(ByteBuffer buffer) {
        System.out.println("mark：" + buffer.mark());
        System.out.println("position：" + buffer.position());
        System.out.println("limit：" + buffer.limit());
        System.out.println("capacity：" + buffer.capacity());
    }
}

public class ByteBufDemo {

    public static void main(String[] args) {
        // 创建一个缓冲区
        ByteBuf buffer = Unpooled.buffer(10);
        System.out.println("------------初始时缓冲区------------");
        printBuffer(buffer);

        // 添加一些数据到缓冲区中
        System.out.println("------------添加数据到缓冲区------------");
        String s = "love";
        buffer.writeBytes(s.getBytes());
        printBuffer(buffer);

        // 读取数据
        System.out.println("------------读取数据------------");
        while (buffer.isReadable()) {
            System.out.println(buffer.readByte());
        }
        printBuffer(buffer);

        // 执行 compact
        System.out.println("------------执行discardReadBytes------------");
        buffer.discardReadBytes();
        printBuffer(buffer);

        // 执行clear
        System.out.println("------------执行clear清空缓冲区------------");
        buffer.clear();
        printBuffer(buffer);
    }

    /**
     * 打印出ByteBuf的信息
     *
     * @param buffer
     */
    private static void printBuffer(ByteBuf buffer) {
        System.out.println("readerIndex：" + buffer.readerIndex());
        System.out.println("writerIndex：" + buffer.writerIndex());
        System.out.println("capacity：" + buffer.capacity());
    }
}

public class ByteBufHeapBufferDemo {

    public static void main(String[] args) {

        // 创建一个堆缓冲区
        ByteBuf buffer = Unpooled.buffer(10);
        String s = "waylau";
        buffer.writeBytes(s.getBytes());

        // 检查是否是支撑数组
        if (buffer.hasArray()) {
            // 获取支撑数据的引用
            byte[] array = buffer.array();
            // 计算第一个字节的偏移量
            int offset = buffer.readerIndex() + buffer.arrayOffset();
            // 可读字节数
            int length = buffer.readableBytes();
            printBuffer(array, offset, length);
        }
    }

    /**
     * 打印出Buffer的信息
     *
     * @param buffer
     */
    private static void printBuffer(byte[] array, int offset, int len) {
        System.out.println("array：" + array);
        System.out.println("array->String：" + new String(array));
        System.out.println("offset：" + offset);
        System.out.println("len：" + len);
    }
}

public class ByteBufDirectBufferDemo {

    public static void main(String[] args) {
        ByteBuf buffer = Unpooled.directBuffer(10);
        String s = "waylau";
        buffer.writeBytes(s.getBytes());

        // 检查是否是支撑数组.
        // 不是支撑数组，则为直接缓冲区
        if (!buffer.hasArray()) {
            // 计算第一个字节的偏移量
            int offset = buffer.readerIndex();
            // 可读字节数
            int length = buffer.readableBytes();
            // 获取字节内容
            byte[] array = new byte[length];
            buffer.getBytes(offset, array);

            printBuffer(array, offset, length);
        }
    }

    /**
     * 打印出Buffer的信息
     *
     * @param buffer
     */
    private static void printBuffer(byte[] array, int offset, int len) {
        System.out.println("array：" + array);
        System.out.println("array->String：" + new String(array));
        System.out.println("offset：" + offset);
        System.out.println("len：" + len);
    }
}

public class ByteBufCompositeBufferDemo {

    public static void main(String[] args) {
        // 创建一个堆缓冲区
        ByteBuf heapBuf = Unpooled.buffer(3);
        String way = "way";
        heapBuf.writeBytes(way.getBytes());

        // 创建一个直接缓冲区
        ByteBuf directBuf = Unpooled.directBuffer(3);
        String lau = "lau";
        directBuf.writeBytes(lau.getBytes());

        // 创建一个复合缓冲区
        CompositeByteBuf compositeBuffer = Unpooled.compositeBuffer(10);
        // 将缓冲区添加到符合缓冲区
        compositeBuffer.addComponents(heapBuf,directBuf);

        // 检查是否是支撑数组.
        // 不是支撑数组，则为复合缓冲区
        if (!compositeBuffer.hasArray()) {
            for (ByteBuf buffer : compositeBuffer) {
                // 计算第一个字节的偏移量
                int offset = buffer.readerIndex();
                // 可读字节数
                int length = buffer.readableBytes();
                // 获取字节内容
                byte[] array = new byte[length];
                buffer.getBytes(offset, array);
                printBuffer(array, offset, length);
            }
        }
    }

    /**
     * 打印出Buffer的信息
     *
     * @param buffer
     */
    private static void printBuffer(byte[] array, int offset, int len) {
        System.out.println("array：" + array);
        System.out.println("array->String：" + new String(array));
        System.out.println("offset：" + offset);
        System.out.println("len：" + len);
    }
}

public class DatagramChannelEchoServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 消息转为DatagramPacket类型
        DatagramPacket packet = (DatagramPacket) msg;
        System.out.println(packet.sender() + " -> Server :" + msg);

        /**
         * 构建新DatagramPacket
         * 将新的 DatagramPacket的接收者设置为packet 的发送者，实现将接收到的客户端消息发回给客户端
         */
        DatagramPacket data = new DatagramPacket(packet.content(), packet.sender());

        // 写消息到管道
        ctx.write(data);// 写消息
        ctx.flush(); // 刷新消息

        // 上面两个方法等同于 ctx.writeAndFlush(data);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        // 当出现异常就关闭连接
        cause.printStackTrace();
        ctx.close();
    }
}

public class DatagramChannelEchoServer {

    public static int DEFAULT_PORT = 7;

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

        try {
            port = Integer.parseInt(args[0]);
        } catch (RuntimeException ex) {
            port = DEFAULT_PORT;
        }

        // 配置事件循环器
        // UDP 只需要一个
        EventLoopGroup group = new NioEventLoopGroup();

        try {
            // 启动NIO服务的引导程序类
            Bootstrap b = new Bootstrap();

            b.group(group) // 设置EventLoopGroup
                    .channel(NioDatagramChannel.class) // 指明新的Channel的类型
                    .option(ChannelOption.SO_BROADCAST, true) // 设置的Channel的一些选项 SO_BROADCAST 发送广播消息
                    .handler(new DatagramChannelEchoServerHandler()); // 指定ChannelHandler

            // 绑定端口
            ChannelFuture f = b.bind(port).sync();

            System.out.println("DatagramChannelEchoServer已启动，端口：" + port);

            // 等待服务器 socket 关闭 。
            // 在这个例子中，这不会发生，但你可以优雅地关闭你的服务器。
            f.channel().closeFuture().sync();
        } finally {

            // 优雅的关闭
            group.shutdownGracefully();
        }
    }
}

public class DatagramChannelEchoClientHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {

        // 从管道读消息
        DatagramPacket packet = (DatagramPacket) msg; // 转为DatagramPacket类型
        String m = packet.content().toString(CharsetUtil.UTF_8);  // 转为字符串
        System.out.println("echo :" + m);
    }

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

        // 当出现异常就关闭连接
        cause.printStackTrace();
        ctx.close();
    }
}

public class DatagramChannelEchoClient {

    public static void main(String[] args) throws InterruptedException {
        if (args.length != 2) {
            System.err.println("用法: java DatagramChannelEchoClient <host name> <port number>");
            System.exit(1);
        }
        String host = args[0];
        int port = Integer.parseInt(args[1]);

        // 配置客户端
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                    .channel(NioDatagramChannel.class)
                    .option(ChannelOption.SO_BROADCAST, true)	// 广播消息
                    .handler(new DatagramChannelEchoClientHandler());

            // 绑定端口
            ChannelFuture f = b.bind(port).sync();
            System.out.println("DatagramChannelEchoClient已启动，端口：" + port);

            Channel channel = f.channel();
            ByteBuffer writeBuffer = ByteBuffer.allocate(32);
            try (BufferedReader stdIn = new BufferedReader(new InputStreamReader(System.in))) {
                String userInput;
                while ((userInput = stdIn.readLine()) != null) {
                    writeBuffer.put(userInput.getBytes());
                    writeBuffer.flip();
                    writeBuffer.rewind();

                    // 转为ByteBuf
                    ByteBuf buf = Unpooled.copiedBuffer(writeBuffer);
                    // 写消息到管道
                    // 消息封装为DatagramPacket类型
                    channel.writeAndFlush(new DatagramPacket(buf,
                            new InetSocketAddress(host, DatagramChannelEchoServer.DEFAULT_PORT)));

                    // 清理缓冲区
                    writeBuffer.clear();
                }
            } catch (UnknownHostException e) {
                System.err.println("不明主机，主机名为： " + host);
                System.exit(1);
            } catch (IOException e) {
                System.err.println("不能从主机中获取I/O，主机名为：" + host);
                System.exit(1);
            }
        } finally {
            // 优雅的关闭
            group.shutdownGracefully();
        }
    }
}

public class Reactor implements Runnable {

	private final Selector selector;
	private final ServerSocketChannel serverSocketChannel;

	public Reactor(int port) throws IOException {
		selector = Selector.open(); // 打开一个Selector
		serverSocketChannel = ServerSocketChannel.open(); // 建立一个Server端通道
		serverSocketChannel.socket().bind(new InetSocketAddress(port)); // 绑定服务端口
		serverSocketChannel.configureBlocking(false); // 置为非阻塞模式

		// Reactor是入口，最初给一个channel注册上去的事件都是accept
		SelectionKey sk = serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

		// 绑定Acceptor处理类
		sk.attach(new Acceptor(serverSocketChannel));
	}

	@Override
	public void run() {
		try {
			while (!Thread.interrupted()) {
				int count = selector.select(); // 就绪事件到达之前，阻塞
				if (count == 0) {
					continue;
				}
				Set<SelectionKey> selected = selector.selectedKeys(); // 拿到本次select获取的就绪事件
				Iterator<SelectionKey> it = selected.iterator();
				while (it.hasNext()) {
					// 这里进行任务分发
					dispatch((SelectionKey) (it.next()));
				}
				selected.clear();
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	void dispatch(SelectionKey k) {
		// 附带对象为Acceptor
		Runnable r = (Runnable) (k.attachment());

		// 调用之前注册的回调对象
		if (r != null) {
			r.run();
		}
	}
}

public class Acceptor implements Runnable {

	private final ServerSocketChannel serverSocketChannel;

	private final int coreNum = Runtime.getRuntime().availableProcessors(); // CPU核心数

	private final Selector[] selectors = new Selector[coreNum]; // 创建selector给SubReactor使用

	private int next = 0; // 轮询使用subReactor的下标索引

	private SubReactor[] reactors = new SubReactor[coreNum]; // subReactor

	private Thread[] threads = new Thread[coreNum]; // subReactor的处理线程

	Acceptor(ServerSocketChannel serverSocketChannel) throws IOException {
		this.serverSocketChannel = serverSocketChannel;
		// 初始化
		for (int i = 0; i < coreNum; i++) {
			selectors[i] = Selector.open();
			reactors[i] = new SubReactor(selectors[i], i); // 初始化sub reactor
			threads[i] = new Thread(reactors[i]); // 初始化运行sub reactor的线程
			threads[i].start(); // 启动（启动后的执行参考SubReactor里的run方法）
		}
	}

	@Override
	public void run() {
		SocketChannel socketChannel;
		try {
			socketChannel = serverSocketChannel.accept(); // 连接
			if (socketChannel != null) {
				System.out.println(String.format("accpet %s", socketChannel.getRemoteAddress()));
				socketChannel.configureBlocking(false);

				// 注意一个selector在select时是无法注册新事件的，因此这里要先暂停下select方法触发的程序段，
				// 下面的weakup和这里的setRestart都是做这个事情的，具体参考SubReactor里的run方法
				reactors[next].registering(true);
				selectors[next].wakeup(); // 使一个阻塞住的selector操作立即返回
				SelectionKey selectionKey = 
						socketChannel.register(selectors[next], SelectionKey.OP_READ); // 注册一个读事件
				selectors[next].wakeup(); // 使一个阻塞住的selector操作立即返回

				// 本次事件注册完成后，需要再次触发select的执行，
				// 因此这里Restart要在设置回false（具体参考SubReactor里的run方法）
				reactors[next].registering(false);

				// 绑定Handler
				selectionKey.attach(new AsyncHandler(socketChannel, selectors[next], next));
				if (++next == selectors.length) {
					next = 0; // 越界后重新分配
				}
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
}

public class SubReactor implements Runnable {
	private final Selector selector;
	private boolean register = false; // 注册开关表示
	private int num; // 序号，也就是Acceptor初始化SubReactor时的下标

	SubReactor(Selector selector, int num) {
		this.selector = selector;
		this.num = num;
	}

	@Override
	public void run() {
		while (!Thread.interrupted()) {
			System.out.println(String.format("NO %d SubReactor waitting for register...", num));
			while (!Thread.interrupted() && !register) {
				try {
					if (selector.select() == 0) {
						continue;
					}
				} catch (IOException e) {
					e.printStackTrace();
				}
				Set<SelectionKey> selectedKeys = selector.selectedKeys();
				Iterator<SelectionKey> it = selectedKeys.iterator();
				while (it.hasNext()) {
					dispatch(it.next());
					it.remove();
				}
			}
		}
	}

	private void dispatch(SelectionKey key) {
		Runnable r = (Runnable) (key.attachment());
		if (r != null) {
			r.run();
		}
	}

	void registering(boolean register) {
		this.register = register;
	}

}

public class AsyncHandler implements Runnable {

	private final Selector selector;

	private final SelectionKey selectionKey;
	private final SocketChannel socketChannel;

	private ByteBuffer readBuffer = ByteBuffer.allocate(1024);
	private ByteBuffer sendBuffer = ByteBuffer.allocate(2048);

	private final static int READ = 0; // 读取就绪
	private final static int SEND = 1; // 响应就绪
	private final static int PROCESSING = 2; // 处理中

	private int status = READ; // 所有连接完成后都是从一个读取动作开始的

	private int num; // 从反应堆序号

	// 开启线程数为4的异步处理线程池
	private static final ExecutorService workers = Executors.newFixedThreadPool(5);

	AsyncHandler(SocketChannel socketChannel, Selector selector, int num) throws IOException {
		this.num = num; // 为了区分Handler被哪个从反应堆触发执行做的标记
		this.socketChannel = socketChannel; // 接收客户端连接
		this.socketChannel.configureBlocking(false); // 置为非阻塞模式
		selectionKey = socketChannel.register(selector, 0); // 将该客户端注册到selector
		selectionKey.attach(this); // 附加处理对象，当前是Handler对象
		selectionKey.interestOps(SelectionKey.OP_READ); // 连接已完成，那么接下来就是读取动作
		this.selector = selector;
		this.selector.wakeup();
	}

	@Override
	public void run() {
		// 如果一个任务正在异步处理，那么这个run是直接不触发任何处理的，
		// read和send只负责简单的数据读取和响应，业务处理完全不阻塞这里的处理
		switch (status) {
		case READ:
			read();
			break;
		case SEND:
			send();
			break;
		default:
		}
	}

	private void read() {
		if (selectionKey.isValid()) {
			try {
				readBuffer.clear();

				// read方法结束，意味着本次"读就绪"变为"读完毕"，标记着一次就绪事件的结束
				int count = socketChannel.read(readBuffer);
				if (count > 0) {
					status = PROCESSING; // 置为处理中
					workers.execute(this::readWorker); // 异步处理
				} else {
					selectionKey.cancel();
					socketChannel.close();
					System.out.println(String.format("NO %d SubReactor read closed", num));
				}
			} catch (IOException e) {
				System.err.println("处理read业务时发生异常！异常信息：" + e.getMessage());
				selectionKey.cancel();
				try {
					socketChannel.close();
				} catch (IOException e1) {
					System.err.println("处理read业务关闭通道时发生异常！异常信息：" + e.getMessage());
				}
			}
		}
	}

	void send() {
		if (selectionKey.isValid()) {
			status = PROCESSING; // 置为执行中
			workers.execute(this::sendWorker); // 异步处理
			selectionKey.interestOps(SelectionKey.OP_READ); // 重新设置为读
		}
	}

	// 读入信息后的业务处理
	private void readWorker() {
		try {

			// 模拟一段耗时操作
			Thread.sleep(5000L);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		try {
			System.out.println(String.format("NO %d %s -> Server： %s", 
					num, socketChannel.getRemoteAddress(),
					new String(readBuffer.array())));
		} catch (IOException e) {
			System.err.println("异步处理read业务时发生异常！异常信息：" + e.getMessage());
		}
		status = SEND;
		selectionKey.interestOps(SelectionKey.OP_WRITE); // 注册写事件
		this.selector.wakeup(); // 唤醒阻塞在select的线程
	}

	private void sendWorker() {
		try {
			sendBuffer.clear();
			sendBuffer.put(String.format("NO %d SubReactor recived %s from %s", num,
					new String(readBuffer.array()), 
					socketChannel.getRemoteAddress()).getBytes());
			sendBuffer.flip();

			// write方法结束，意味着本次写就绪变为写完毕，标记着一次事件的结束
			int count = socketChannel.write(sendBuffer);

			if (count < 0) {
				// 同上，write场景下，取到-1，也意味着客户端断开连接
				selectionKey.cancel();
				socketChannel.close();
				System.out.println(String.format("%d SubReactor send closed", num));
			}

			// 没断开连接，则再次切换到读
			status = READ;
		} catch (IOException e) {
			System.err.println("异步处理send业务时发生异常！异常信息：" + e.getMessage());
			selectionKey.cancel();
			try {
				socketChannel.close();
			} catch (IOException e1) {
				System.err.println("异步处理send业务关闭通道时发生异常！异常信息：" + e.getMessage());
			}
		}
	}
}

public class MainSubReactorDemo {

    public static void main(String[] args) throws IOException {
        new Thread(new Reactor(2333)).start();
    }

}

EventLoopGroup group = new NioEventLoopGroup(1); // 线程为1
ServerBootstrap b = new ServerBootstrap();

EventLoopGroup group = new NioEventLoopGroup(); // 不指定线程数，则为cpu 核数
ServerBootstrap b = new ServerBootstrap();

 // 多事件循环器
 EventLoopGroup bossGroup = new NioEventLoopGroup(); // boss
 EventLoopGroup workerGroup = new NioEventLoopGroup(); // worker

public class SquareDecoder extends ByteToMessageDecoder {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        out.add(in.readBytes(in.readableBytes()));
    }
}

public class ToIntegerDecoder2 extends ReplayingDecoder {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        out.add(in.readInt());
    }
}

public class IntegerHeaderFrameDecoder extends ReplayingDecoder<Void> {

    private boolean readLength;
    private int length;

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf buf, List<Object> out) throws Exception {
        if (!readLength) {
            length = buf.readInt();
            readLength = true;
            checkpoint();
        }
        if (readLength) {
            ByteBuf frame = buf.readBytes(length);
            readLength = false;
            checkpoint();
            out.add(frame);
        }
    }
}

public class IntegerHeaderFrameDecoder2 extends ReplayingDecoder<MyDecoderState> {

    private int length;

    public IntegerHeaderFrameDecoder2() {
        // 设置初始状态
        super(MyDecoderState.READ_LENGTH);
    }

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf buf, List<Object> out) throws Exception {
        switch (state()) {
            case READ_LENGTH:
                length = buf.readInt();
                checkpoint(MyDecoderState.READ_CONTENT);
            case READ_CONTENT:
                ByteBuf frame = buf.readBytes(length);
                checkpoint(MyDecoderState.READ_LENGTH);
                out.add(frame);
                break;
            default:
                throw new Error("Should`t reach here.");
        }
    }
}

public class IntegerToStringDecoder extends MessageToMessageDecoder<Integer> {
    @Override
    protected void decode(ChannelHandlerContext ctx, Integer msg, List<Object> out) throws Exception {
        out.add(String.valueOf(msg));
    }
}

public class SafeByteMessageDecoder extends ByteToMessageDecoder {  // 继承主要用于将字节解码为消息
    private static final int MAX_FRAME_SIZE = 1024;

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        int readbale = in.readableBytes();
        if (readbale > MAX_FRAME_SIZE) {    // 检测缓冲区数据是否大于常量 MAX_FRAME_SIZE
            // 忽略所有可读的字节，并抛出 TooLongFrameException 来通知ChannelPipeline 中的ChannelHander
            // 以说明这个帧数据超长
            in.skipBytes(readbale);
            throw new TooLongFrameException("Frame too big !");
        }
    }
}

public class MyLineBasedFrameDecoder extends LineBasedFrameDecoder {

    private final static int MAX_LENGTH = 1024; // 帧的最大长度

    public MyLineBasedFrameDecoder() {
        super(MAX_LENGTH);
    }

}

public class MyLineBasedFrameDecoderServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        // 接收msg消息，此处已经无需解码了
        System.out.println("Client -> Server: " + msg);
    }
}

public class MyLineBasedFrameDecoderChannelInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel channel) throws Exception {
        // 基于换行符号
        channel.pipeline().addLast(new MyLineBasedFrameDecoder());
        // 解码转String
        channel.pipeline().addLast(new StringDecoder());
        // 自定义ChannelHandler
        channel.pipeline().addLast(new MyLineBasedFrameDecoderServerHandler());
    }
}

public class MyLineBasedFrameDecoderServer {

    public static int DEFAULT_PORT = 8023;

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

        try {
            port = Integer.parseInt(args[0]);
        } catch (RuntimeException ex) {
            port = DEFAULT_PORT;
        }

        // 多线程事件循环器
        EventLoopGroup bossGroup = new NioEventLoopGroup(1); // boss
        EventLoopGroup workerGroup = new NioEventLoopGroup(); // worker

        try {
            // 启动NIO服务的引导程序类
            ServerBootstrap b = new ServerBootstrap();

            b.group(bossGroup, workerGroup) // 设置EventLoopGroup
                    .channel(NioServerSocketChannel.class) // 指明新的Channel的类型
                    .childHandler(new MyLineBasedFrameDecoderChannelInitializer()) // 指定ChannelHandler
                    .option(ChannelOption.SO_BACKLOG, 128) // 设置的ServerChannel的一些选项
                    .childOption(ChannelOption.SO_KEEPALIVE, true); // 设置的ServerChannel的子Channel的选项

            // 绑定端口，开始接收进来的连接
            ChannelFuture f = b.bind(port).sync();

            System.out.println("MyLineBasedFrameDecoderServer已启动，端口：" + port);

            // 等待服务器 socket 关闭 。
            // 在这个例子中，这不会发生，但你可以优雅地关闭你的服务器。
            f.channel().closeFuture().sync();
        } finally {
            // 优雅的关闭
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }
}

public class TcpClient {

    public static void main(String[] args) throws IOException {
        Socket socket = null;
        OutputStream out = null;

        try {

            socket = new Socket("localhost", 8023);
            out = socket.getOutputStream();

            // 请求服务器
            String lines = "床前明月光\r\n疑是地上霜\r\n举头望明月\r\n低头思故乡\r\n";
            byte[] outputBytes = lines.getBytes("UTF-8");
            out.write(outputBytes);
            out.flush();
        } finally {
            // 关闭连接
            out.close();
            socket.close();
        }
    }
}

public class ShortToByteEncoder extends MessageToByteEncoder<Short> {
    @Override
    protected void encode(ChannelHandlerContext ctx, Short msg, ByteBuf out) throws Exception {
        out.writeShort(msg);
    }
}

public class IntegerToStringEncoder extends MessageToMessageEncoder {
    @Override
    protected void encode(ChannelHandlerContext ctx, Object msg, List out) throws Exception {
        out.add(String.valueOf(msg));
    }
}

public enum MsgType {

    // 请求
    EMGW_LOGIN_REQ((byte) 0x00),
    // 响应
    EMGW_LOGIN_RES((byte) 0x01);

    private byte value;

    public byte getValue() {
        return value;
    }

    private MsgType(byte value) {
        this.value = value;
    }
}

public class MsgHeader {
    private byte msgType; // 消息类型
    private int len; // 长度

    public MsgHeader() {
    }

    public MsgHeader(byte msgType, int len) {
        this.msgType = msgType;
        this.len = len;
    }

    public byte getMsgType() {
        return msgType;
    }

    public void setMsgType(byte msgType) {
        this.msgType = msgType;
    }

    public int getLen() {
        return len;
    }

    public void setLen(int len) {
        this.len = len;
    }
}

public class Msg {

    private MsgHeader msgHeader = new MsgHeader();
    private String body;

    public MsgHeader getMsgHeader() {
        return msgHeader;
    }

    public void setMsgHeader(MsgHeader msgHeader) {
        this.msgHeader = msgHeader;
    }

    public String getBody() {
        return body;
    }

    public void setBody(String body) {
        this.body = body;
    }
}

/**
 * 编码器
 * @author: wangyj
 * @create: 2022-04-30
 * @version: 1.0.0
 **/
public class MyEncoder extends MessageToByteEncoder<Msg> {
    @Override
    protected void encode(ChannelHandlerContext ctx, Msg msg, ByteBuf out) throws Exception {
        if (msg == null | msg.getMsgHeader() == null) {
            throw new Exception("The encode message is null");
        }

        // 获取消息头
        MsgHeader header = msg.getMsgHeader();
        // 获取消息体
        String body = msg.getBody();
        byte[] bodyBytes = body.getBytes(Charset.forName("utf-8"));
        // 计算消息体的长度
        int bodySize = bodyBytes.length;
        System.out.printf("MyEncoder header: %s, body: %s", header.getMsgType(), body);

        out.writeByte(MsgType.EMGW_LOGIN_RES.getValue());
        out.writeInt(bodySize);
        out.writeBytes(bodyBytes);
    }
}

public class MyDecoder extends LengthFieldBasedFrameDecoder {

    private static final int MAX_FRAME_LENGTH = 1024 * 1024;
    private static final int LENGTH_FIELD_LENGTH = 4;
    private static final int LENGTH_FIELD_OFFSET = 1;
    private static final int LENGTH_ADJUSTMENT = 0;
    private static final int INITIAL_BYTES_TO_STRIP = 0;

    private static final int HEADER_SIZE = 5;
    private byte msgType; // 消息类型
    private int len; // 长度

    public MyDecoder() {
        super(MAX_FRAME_LENGTH,
                LENGTH_FIELD_OFFSET, LENGTH_FIELD_LENGTH,
                LENGTH_ADJUSTMENT, INITIAL_BYTES_TO_STRIP);
    }

    @Override
    protected Object decode(ChannelHandlerContext ctx, ByteBuf in2) throws Exception {
        ByteBuf in = (ByteBuf) super.decode(ctx, in2);
        if (in == null) {
            return null;
        }
        // 校验头长度
        if (in.readableBytes() < HEADER_SIZE) {
            return null;
        }
        msgType = in.readByte();
        len = in.readInt();

        // 校验消息体长度
        if (in.readableBytes() < len) {
            return null;
        }

        ByteBuf buf = in.readBytes(len);
        byte[] req = new byte[buf.readableBytes()];
        buf.readBytes(req);
        String body = new String(req, "UTF-8");

        // ByteBuf转为Msg类型
        Msg msg = new Msg();
        MsgHeader msgHeader = new MsgHeader(msgType, len);
        msg.setBody(body);
        msg.setMsgHeader(msgHeader);
        return msg;
    }
}

public class MyServerHandler extends SimpleChannelInboundHandler {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object obj) throws Exception {
        Channel incoming = ctx.channel();

        if (obj instanceof Msg) {
            Msg msg = (Msg) obj;
            System.out.println("Client->Server:" + incoming.remoteAddress() + msg.getBody());
            incoming.write(obj);
        }
    }
}

public class MyServer {

    private int port;

    public MyServer(int port) {
        this.port = port;
    }

    private void run() throws InterruptedException {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            // 这里需要注意添加的顺序
                            ch.pipeline().addLast("decoder", new MyDecoder());
                            ch.pipeline().addLast("encoder", new MyEncoder());
                            ch.pipeline().addLast(new MyServerHandler());
                        }
                    }).option(ChannelOption.SO_BACKLOG, 128)
                    .childOption(ChannelOption.SO_KEEPALIVE, true);

            ChannelFuture f = b.bind(port).sync();
            System.out.println("Server start listen at " + port);

            f.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception {
        int port;
        if (args.length > 0) {
            port = Integer.parseInt(args[0]);
        } else {
            port = 8082;
        }
        new MyServer(port).run();
    }
}

public class MyClient {

    private String host;
    private int port;

    public MyClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

    private void run() throws InterruptedException {
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(workerGroup);
            b.channel(NioSocketChannel.class);
            b.option(ChannelOption.SO_KEEPALIVE, true);
            b.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast("decoder", new MyDecoder());
                    ch.pipeline().addLast("encoder", new MyEncoder());
                    ch.pipeline().addLast(new MyClientHandler());

                }
            });

            // 启动客户端
            ChannelFuture f = b.connect(host, port).sync();

            while (true) {
                Msg msg = new Msg();
                MsgHeader msgHeader = new MsgHeader();
                msgHeader.setMsgType(MsgType.EMGW_LOGIN_REQ.getValue());
                String body = "床前明月光，疑是地上霜。举头望明月，低头思故乡。";

                byte[] bodyBytes = body.getBytes(Charset.forName("utf-8"));
                int bodySize = bodyBytes.length;
                msgHeader.setLen(bodySize);
                msg.setMsgHeader(msgHeader);
                msg.setBody(body);

                f.channel().writeAndFlush(msg);
                Thread.sleep(2000);
            }
        } finally {
            workerGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        new MyClient("localhost", 8082).run();
    }
}

public class MyCodec extends CombinedChannelDuplexHandler<MyDecoder, MyEncoder> {
    public MyCodec() {
        super(new MyDecoder(), new MyEncoder());
    }
}

b.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    // 添加自己的编解码器
                    ch.pipeline().addLast("codec", new MyCodec());
                    ch.pipeline().addLast(new MyClientHandler());

                }
            });

public class SerializationBean implements Serializable {

	private static final long serialVersionUID = 3235432002462705915L;
	private int age;
	private String name;

	public int getAge() {
		return age;
	}

	public void setAge(int age) {
		this.age = age;
	}

	public String getName() {
		return name;
	}

	public void setName(String name) {
		this.name = name;
	}

	@Override
	public String toString() {
		return "SerializationBean [age=" + age + ", name=" + name + "]";
	}
}

public class SerializationServerHandler extends SimpleChannelInboundHandler<Object> {

	@Override
	protected void channelRead0(ChannelHandlerContext ctx, Object obj) throws Exception {
		if (obj instanceof SerializationBean) {
			SerializationBean user = (SerializationBean) obj;
			ctx.writeAndFlush(user);
			System.out.println("Client -> Server: " + user);
		}
	}
}

public class SerializationServerInitializer extends ChannelInitializer<Channel> {

	private final static int MAX_OBJECT_SIZE = 1024 * 1024;

	@Override
	protected void initChannel(Channel ch) throws Exception {
		ChannelPipeline pipeline = ch.pipeline();
		pipeline.addLast(new ObjectDecoder(MAX_OBJECT_SIZE,
				ClassResolvers.weakCachingConcurrentResolver(this.getClass()
						.getClassLoader())));
		pipeline.addLast(new ObjectEncoder());
		pipeline.addLast(new SerializationServerHandler());
	}
}

public final class SerializationServer {

    static final int PORT = 8082;

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

        // Configure the server.
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .option(ChannelOption.SO_BACKLOG, 100)
             .childOption(ChannelOption.SO_KEEPALIVE, true)
             .handler(new LoggingHandler(LogLevel.INFO))
             .childHandler(new SerializationServerInitializer());

            // Start the server.
            ChannelFuture f = b.bind(PORT).sync();

            // Wait until the server socket is closed.
            f.channel().closeFuture().sync();
        } finally {
            // Shut down all event loops to terminate all threads.
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}