public class MyMessageEncoder extends MessageToByteEncoder<MessagePO> {
    @Override
    protected void encode(ChannelHandlerContext ctx, MessagePO msg, ByteBuf out) throws Exception {
        System.out.println("MyMessageEncoder.encode，被調(diào)用");
        String json = JSONObject.toJSONString(msg);
        out.writeInt(json.getBytes(StandardCharsets.UTF_8).length);
        out.writeBytes(json.getBytes(StandardCharsets.UTF_8));
    }
}

public class MyMessageDecoder extends ByteToMessageDecoder {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        System.out.println("MyMessageDecoder.decode，被調(diào)用");
        while (in.readableBytes() >= 4){
            int num = in.readInt();
            System.out.println("解碼出一個整數(shù)："+num);
            out.add(num);
        }
    }
}

public class MyMessageDecoder extends ReplayingDecoder<Void> {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        System.out.println("MyMessageDecoder.decode，被調(diào)用");
        int length = in.readInt();
        byte[] content = new byte[length];
        in.readBytes(content);
        String json = new String(content,StandardCharsets.UTF_8);
        MessagePO po = JSONObject.parseObject(json,MessagePO.class);
        out.add(po);
    }
}

public class RequestMessageCodec extends MessageToMessageCodec<String, RequestData> {
    @Override
    protected void encode(ChannelHandlerContext ctx, RequestData msg, List<Object> out) throws Exception {
        System.out.println("RequestMessageCodec.encode 被調(diào)用 " + msg);
        String json = JSONObject.toJSONString(msg);
        out.add(json);
    }
    @Override
    protected void decode(ChannelHandlerContext ctx, String msg, List<Object> out) throws Exception {
        System.out.println("RequestMessageCodec.decode 被調(diào)用 " + msg);
        RequestData po = JSONObject.parseObject(msg, RequestData.class);
        out.add(po);
    }
}

Bootstrap bootstrap = new Bootstrap().group(group)//設(shè)置線程組
    .channel(NioSocketChannel.class)//設(shè)置客戶端通道的實現(xiàn)類
    .handler(new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ch.pipeline().addLast(new NettyClientHandler());//加入自己的處理器
        }
    });

ServerBootstrap bootstrap = new ServerBootstrap().group(bossGroup, workerGroup)
    .channel(NioServerSocketChannel.class)//使用NioServerSocketChannel作為服務(wù)器的通道實現(xiàn)
    .option(ChannelOption.SO_BACKLOG, 128)//設(shè)置線程隊列等待連接的個數(shù)
    .childOption(ChannelOption.SO_KEEPALIVE, true)//設(shè)置保持活動連接狀態(tài)
//      .handler(null)//該Handler對應(yīng)bossGroup
    .childHandler(new ChannelInitializer<SocketChannel>() {//給workerGroup的EventLoop對應(yīng)的管道設(shè)置處理器
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ch.pipeline().addLast(new NettyServerHandler());
        }
    });

public class MyClientHandler extends SimpleChannelInboundHandler<MessagePO> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, MessagePO msg) throws Exception {
        System.out.println("收到服務(wù)端消息：" + msg);
    }
}

DefaultEventLoopGroup businessGroup = new DefaultEventLoopGroup(100);
...
addLast(businessGroup, new MyNettyServerHandler())

/**
 * 讀取客戶端發(fā)送過來的消息
 */
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    ByteBuf byteBuf = (ByteBuf) msg;
    System.out.println("收到客戶信息：" + byteBuf.toString(CharsetUtil.UTF_8));
    System.out.println("客戶端地址：" + ctx.channel().remoteAddress());
    System.out.println("處理線程：" + Thread.currentThread().getName());
    ctx.executor().parent().execute(()->{
        try {
            System.out.println("parent().execute Thread = " + Thread.currentThread().getName());
            TimeUnit.SECONDS.sleep(2L);
            System.out.println("parent任務(wù)執(zhí)行完成1");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    });
    ctx.executor().parent().execute(()->{
        try {
            System.out.println("parent().execute Thread = " + Thread.currentThread().getName());
            TimeUnit.SECONDS.sleep(2L);
            System.out.println("parent任務(wù)執(zhí)行完成2");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    });
    ctx.executor().parent().execute(()->{
        try {
            System.out.println("parent().execute Thread = " + Thread.currentThread().getName());
            TimeUnit.SECONDS.sleep(2L);
            System.out.println("parent任務(wù)執(zhí)行完成3");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    });
}

ctx.channel().eventLoop().schedule(()->{
    try {
        System.out.println("Thread.currentThread().getName() = " + Thread.currentThread().getName());
        TimeUnit.SECONDS.sleep(2L);
        System.out.println("定時任務(wù)執(zhí)行完成");
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
},10L,TimeUnit.SECONDS);