服务端代码

服务端主启动类

public class MyServer {
    public static void main(String[] args) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            /*
                所有的Handler() 都是针对bossGroup发挥作用
                所有的childHandler() 都是针对workerGroup发挥作用，连接丢给workerGroup
                    之后由childHandler对象对workerGroup这个Nio线程发挥作用
             */
            serverBootstrap.group(bossGroup,workerGroup).channel(NioServerSocketChannel.class)
                    .childHandler(new MyServerInitializer());
            ChannelFuture channelFuture = serverBootstrap.bind(8899).sync();
            channelFuture.channel().closeFuture().sync();
        }finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

初始化器 (Initializer)

public class MyServerInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        /*
            LengthFieldBasedFrameDecoder 解码器
            LengthFieldPrepender 编码器
        */
        pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE,0,4,0,4));
        pipeline.addLast(new LengthFieldPrepender(4));
        pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
        pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
        pipeline.addLast(new MyServerHandler());

    }
}

自定义处理器 (Handler)

//SimpleChannelInboundHandler<?> 的泛型表示客户端发来的数据 真正的类型是什么
public class MyServerHandler extends SimpleChannelInboundHandler<String> {

    /**
     * 读取客户端发过来的请求，并且向客户端返回响应的方法
     * @param ctx 上下文，可以获取远程的信息，地址、连接对象
     * @param msg 客户端发来的请求对象
     * @throws Exception
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        System.out.println(ctx.channel().remoteAddress() + "," + msg);

        /*
            write 写入缓冲
            flush 把缓冲的内容清/推出去
            writeAndFlush 写入缓冲，并把缓冲的内容推出去
         */
        ctx.channel().writeAndFlush("from server：" + UUID.randomUUID());
    }

    /**
     * 出现异常的情况下怎么办
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

客户端代码

客户端主启动类

public class MyClient {
    public static void main(String[] args) throws Exception {
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class)
                    .handler(new MyClientInitializer());
            //与对应的url建立连接通道
            ChannelFuture channelFuture = bootstrap.connect("localhost",8899).sync();
            channelFuture.channel().closeFuture().sync();
        }finally {
            eventLoopGroup.shutdownGracefully();
        }
    }
}

初始化器 (Initializer)

public class MyClientInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();

        /*
            LengthFieldBasedFrameDecoder 解码器
            LengthFieldPrepender 编码器
        */
        pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE,0,4,0,4));
        pipeline.addLast(new LengthFieldPrepender(4));
        pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
        pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
        pipeline.addLast(new MyClientHandler());
    }
}

自定义处理器 (Handler)

public class MyClientHandler extends SimpleChannelInboundHandler<String> {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        //打印远程地址
        System.out.println(ctx.channel().remoteAddress());

        System.out.println("client output：" + msg);
        ctx.writeAndFlush("from client：" + LocalDateTime.now());
    }

    /**通道活跃时调用*/
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ctx.writeAndFlush("来自于客户端的问候");
    }

    /**发生异常时调用*/
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

运行调试

服务端输出

/127.0.0.1:54554,from client：2020-05-05T17:51:11.729
/127.0.0.1:54554,from client：2020-05-05T17:51:11.729
/127.0.0.1:54554,from client：2020-05-05T17:51:11.730
/127.0.0.1:54554,from client：2020-05-05T17:51:11.730
/127.0.0.1:54554,from client：2020-05-05T17:51:11.730
...

客户端输出

client output：from server：6cf30395-561c-43a6-80d0-429510b3cc84
localhost/127.0.0.1:8899
client output：from server：aa046e14-1535-4b62-9e14-afaa40d371a2
localhost/127.0.0.1:8899
client output：from server：44948915-295a-49b8-89f6-44ce958bb0e2
localhost/127.0.0.1:8899
client output：from server：5a6629d2-114e-4183-a842-2e766be030c0
localhost/127.0.0.1:8899
client output：from server：e519ebe6-417b-4b61-ae1f-efcd65ae7d62
localhost/127.0.0.1:8899
...

流程分析

当客户端和服务端通道建立连接的时候（触发channelActive方法），客户端向服务发送来“自与客户端的问题!”数据，服务器端channelRead0被触发，进行回写数据到客户端触发客户端的channelRead0回调方法被调用，依次重复。