1 public class TimeServerHandler extends ChannelInboundHandlerAdapter{
 2     
 3     /**
 4      * 每个传入的消息都要调用
 5      * */
 6     @Override
 7     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception{
 8         ByteBuf buf = (ByteBuf)msg;
 9         try{
10             byte[] req = new byte[buf.readableBytes()];
11             buf.readBytes(req);
12             String body = new String(req, "UTF-8");
13             System.out.println("服务器收到的请求：" + body);
14             String time = "Query time ?".equals(body) ? new Date().toString() : "fail";
15             ByteBuf resp = Unpooled.copiedBuffer(time.getBytes());
16             ctx.write(resp);
17         }finally{
18             buf.release();
19         }
20     }
21     
22     //通知ChannelInboundHandlerAdapter 最后一次对channelRead()的调用是当前批量读取中的最后一条消息。
23     @Override
24     public void channelReadComplete(ChannelHandlerContext ctx) {
25         /**
26          * ctx.write(resp) 是将信息写到消息队列，而不是每次write就写入SocketChannel
27          * ctx.flush() 是将消息发送队列中的信息写到SocketChannel中发送给对方，
28          * ctx.writeAndFlush(resp) 直接写入SocketChannel
29          * */
30         ctx.flush();
31     }
32     
33     //处理过程中出现的异常
34     @Override
35     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
36         cause.printStackTrace();
37         ctx.close();
38     }
39     
40 }

 1 public class TimeServer {
 2     public void bind(int port) throws Exception{
 3         //NIO线程组,专门用于处理网络事件，本质是Reactor线程组。
 4         EventLoopGroup bosser = new NioEventLoopGroup();//用于服务端接受客户端的连接
 5         EventLoopGroup worker = new NioEventLoopGroup();//用于进行SocketChannel的网络读写
 6         
 7         try {
 8             ServerBootstrap boot = new ServerBootstrap();//启动NIO服务端的辅助启动类，降低开发复杂度。
 9             
10             boot.group(bosser, worker)
11             .channel(NioServerSocketChannel.class)
12             .option(ChannelOption.SO_BACKLOG, 1024)//配置TCP参数
13             .childHandler(new ChannelInitializer<SocketChannel>(){//用于处理网络IO事件（记录日志，对消息进行编解码）
14                 @Override
15                 protected void initChannel(SocketChannel ch) throws Exception {
16                     ch.pipeline().addLast(new TimeServerHandler());
17                 }
18             });
19             
20             ChannelFuture cf = boot.bind(port).sync();//异步地绑定服务器，调用sync()方法阻塞等待直到绑定完成。
21             cf.channel().closeFuture().sync();//获取ChannelFuture，并且阻塞当前线程直到它完成。
22         } finally{
23             bosser.shutdownGracefully();//释放所有资源
24             worker.shutdownGracefully();
25         }
26         
27     }
28     
29     
30     public static void main(String[] args) {
31         int port = 8080;
32         try {
33             new TimeServer().bind(port);
34         } catch (Exception e) {
35             e.printStackTrace();
36         }
37     }
38 }

 1 public class TimeClientHandler extends ChannelInboundHandlerAdapter{
 2     
 3     //在到服务器的连接已经建立之后将被调用
 4     @Override
 5     public void channelActive(ChannelHandlerContext ctx){
 6         byte[] req = "Query time ?".getBytes();
 7         ByteBuf firstMessage = Unpooled.buffer(req.length);
 8         firstMessage.writeBytes(req);
 9         ctx.writeAndFlush(firstMessage);
10     }
11     
12     //当从服务器接受到一条消息时被调用
13     @Override
14     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception{
15         ByteBuf buf = (ByteBuf)msg;
16         byte[] resp = new byte[buf.readableBytes()];
17         buf.readBytes(resp);
18         String str = new String(resp, "UTF-8");
19         System.out.println("服务器时间 : "+ str);
20     }
21     
22     @Override
23     public void channelReadComplete(ChannelHandlerContext ctx) {
24         ctx.flush();
25     }
26  
27     //处理过程中出现的异常
28     @Override
29     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
30         cause.printStackTrace();
31         ctx.close();
32     }
33       
34 }

public class TimeClient {
    public void connect(String host, int port) throws Exception{
        EventLoopGroup group = new NioEventLoopGroup();
        try{
            Bootstrap boot = new Bootstrap();
            boot.group(group).channel(NioSocketChannel.class)
            .option(ChannelOption.TCP_NODELAY, true)
            .handler(new ChannelInitializer<SocketChannel>() {

                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast(new TimeClientHandler());
                }
                
            });
            
            ChannelFuture cf = boot.connect(host, port).sync();
            cf.channel().closeFuture().sync();
        }finally{
            group.shutdownGracefully();
        }
    }
    
    public static void main(String[] args) {
        final String host = "127.0.0.1";
        final int port = 8080;
        try {
            new TimeClient().connect(host, port);
        } catch (Exception e) {
            System.out.println("客户端" + Thread.currentThread().getId() + " 请求出现异常");
        }
                
    }
}