netty 学习

 1 package com.example.demo.nettyLearning;
 2 
 3 import io.netty.bootstrap.Bootstrap;
 4 import io.netty.channel.ChannelFuture;
 5 import io.netty.channel.ChannelInitializer;
 6 import io.netty.channel.EventLoopGroup;
 7 import io.netty.channel.nio.NioEventLoopGroup;
 8 import io.netty.channel.socket.SocketChannel;
 9 import io.netty.channel.socket.nio.NioSocketChannel;
10 
11 import java.net.InetSocketAddress;
12 
13 public class EchoClient {
14 
15     private final String host;
16     private final int port;
17 
18 
19     public EchoClient(String host, int port) {
20         this.host = host;
21         this.port = port;
22     }
23 
24     public static void main(String[] args) throws Exception {
25 
26         final String host ="127.0.0.1";
27         final int port = 6666;
28 
29         new EchoClient(host, port).start();
30     }
31 
32     public void start() throws Exception {
33         EventLoopGroup group = new NioEventLoopGroup();
34         try {
35 
36             Bootstrap b = new Bootstrap();//1.创建 Bootstrap
37             b.group(group)//2.指定 EventLoopGroup 来处理客户端事件。由于我们使用 NIO 传输，所以用到了 NioEventLoopGroup 的实现
38                     .channel(NioSocketChannel.class)//3.使用的 channel 类型是一个用于 NIO 传输
39                     .remoteAddress(new InetSocketAddress(host, port))//4.设置服务器的 InetSocketAddress
40                     .handler(new ChannelInitializer<SocketChannel>() {//5.当建立一个连接和一个新的通道时，创建添加到 EchoClientHandler 实例 到 channel pipeline
41                         @Override
42                         protected void initChannel(SocketChannel ch) throws Exception {
43                             ch.pipeline().addLast(new EchoClientHandler());
44                         }
45                     });
46 
47             ChannelFuture f = b.connect().sync();//6.连接到远程;等待连接完成
48             f.channel().closeFuture().sync();//7.阻塞直到 Channel 关闭
49 
50         } finally {
51 
52             group.shutdownGracefully().sync();//8.调用 shutdownGracefully() 来关闭线程池和释放所有资源
53         }
54     }
55 }

 1 package com.example.demo.nettyLearning;
 2 
 3 import io.netty.buffer.ByteBuf;
 4 import io.netty.buffer.Unpooled;
 5 import io.netty.channel.ChannelHandler;
 6 import io.netty.channel.ChannelHandlerContext;
 7 import io.netty.channel.SimpleChannelInboundHandler;
 8 import io.netty.util.CharsetUtil;
 9 
10 /**
11  * 客户端的工作内容：
12  * 连接服务器
13  * 发送信息
14  * 发送的每个信息，等待和接收从服务器返回的同样的信息
15  * 关闭连接
16  */
17 @ChannelHandler.Sharable //1.@Sharable标记这个类的实例可以在 channel 里共享
18 public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
19 
20 
21     @Override
22     public void channelActive(ChannelHandlerContext ctx) throws Exception {
23         ctx.writeAndFlush(Unpooled.copiedBuffer("Netty rocks!",//2.当被通知该 channel 是活动的时候就发送信息
24                 CharsetUtil.UTF_8));
25     }
26 
27     @Override
28     protected void channelRead0(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
29         System.out.println("Client received: " + in.toString(CharsetUtil.UTF_8));    //3 记录接收到的消息
30     }
31 
32 
33     @Override
34     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {//4.记录日志错误并关闭 channel
35         cause.printStackTrace();
36         ctx.close();
37     }
38 }

 1 package com.example.demo.nettyLearning;
 2 
 3 import io.netty.bootstrap.ServerBootstrap;
 4 import io.netty.channel.ChannelFuture;
 5 import io.netty.channel.ChannelInitializer;
 6 import io.netty.channel.nio.NioEventLoopGroup;
 7 import io.netty.channel.socket.SocketChannel;
 8 import io.netty.channel.socket.nio.NioServerSocketChannel;
 9 
10 public class EchoServer {
11 
12     private final int port;
13 
14     public EchoServer(int port) {
15         this.port = port;
16     }
17 
18 
19     public static void main(String[] args) throws Exception {
20 
21         int port = 6666;//1.设置端口值（抛出一个 NumberFormatException 如果该端口参数的格式不正确）
22         new EchoServer(port).start();//2.呼叫服务器的 start() 方法
23     }
24 
25     public void start() throws Exception {
26 
27         NioEventLoopGroup group = new NioEventLoopGroup();//3.创建 EventLoopGroup
28         try {
29             ServerBootstrap b = new ServerBootstrap();
30             b.group(group)//4.创建 ServerBootstrap
31                     .channel(NioServerSocketChannel.class)//5.指定使用 NIO 的传输 Channel
32                     .localAddress(port)//6.设置 socket 地址使用所选的端口
33                     .childHandler(new ChannelInitializer<SocketChannel>() {//7.添加 EchoServerHandler 到 Channel 的 ChannelPipeline
34 
35                         @Override
36                         protected void initChannel(SocketChannel socketChannel) throws Exception {
37                             socketChannel.pipeline().addLast(new EchoServerHandler());
38                         }
39                     });
40 
41             ChannelFuture f = b.bind().sync();//8.绑定的服务器;sync 等待服务器关闭
42             System.out.println(EchoServer.class.getName() + " started and listen on " + f.channel().localAddress());
43             f.channel().closeFuture().sync();//9.关闭 channel 和 块，直到它被关闭
44 
45         } finally {
46             group.shutdownGracefully().sync();//10.关机的 EventLoopGroup，释放所有资源。
47         }
48     }
49 }

 1 package com.example.demo.nettyLearning;
 2 
 3 import io.netty.buffer.ByteBuf;
 4 import io.netty.buffer.Unpooled;
 5 import io.netty.channel.ChannelFutureListener;
 6 import io.netty.channel.ChannelHandler;
 7 import io.netty.channel.ChannelHandlerContext;
 8 import io.netty.channel.ChannelInboundHandlerAdapter;
 9 import io.netty.util.CharsetUtil;
10 
11 
12 @ChannelHandler.Sharable //@Sharable 标识这类的实例之间可以在 channel 里面共享
13 public class EchoServerHandler extends ChannelInboundHandlerAdapter {
14 
15     @Override
16     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
17         ByteBuf in = (ByteBuf) msg;
18         System.out.println("server received: " + in.toString(CharsetUtil.UTF_8));//日志消息输出到控制台
19         ctx.write(in);//将所接收的消息返回给发送者。注意，这还没有冲刷数据
20     }
21 
22     @Override
23     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
24         ctx.writeAndFlush(Unpooled.EMPTY_BUFFER)//冲刷所有待审消息到远程节点。关闭通道后，操作完成
25                 .addListener(ChannelFutureListener.CLOSE);
26     }
27 
28     @Override
29     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
30         cause.printStackTrace();//打印异常堆栈跟踪
31         ctx.close();//关闭通道
32     }
33 }