Netty 教程

 1import io.netty.bootstrap.ServerBootstrap;
 2import io.netty.channel.*;
 3import io.netty.channel.nio.NioEventLoopGroup;
 4import io.netty.channel.socket.SocketChannel;
 5import io.netty.channel.socket.nio.NioServerSocketChannel;
 6import io.netty.handler.logging.LogLevel;
 7import io.netty.handler.logging.LoggingHandler;
 8
 9import java.net.InetSocketAddress;
10
11/**
12 * echo server，接收打印client发送过来的数据，并把数据返回给client
13 */
14public final class EchoServer {
15
16    static final int PORT = Integer.parseInt(System.getProperty("port", "8007"));
17
18    public static void main(String[] args) throws Exception {
19
20        // 配置服务器
21        // 主event loop，负责接受（accept）socket连接
22        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
23        // 从event loop，负责处理socket的事件
24        EventLoopGroup workerGroup = new NioEventLoopGroup();
25        // server 的channel处理器
26        final EchoServerHandler serverHandler = new EchoServerHandler();
27        try {
28            // Server引导器（配置器）
29            ServerBootstrap b = new ServerBootstrap();
30            b.group(bossGroup, workerGroup)
31                    .channel(NioServerSocketChannel.class)
32                    // 设置ip地址和端口，server默认使用本地ip，也可以通过bind()来设置
33                    .localAddress(new InetSocketAddress(PORT))
34                    // 配置网络传输参数，例如缓存大小
35                    .option(ChannelOption.SO_BACKLOG, 100)
36                    // 配置日志打印
37                    .handler(new LoggingHandler(LogLevel.INFO))
38                    // netty核心组件，pipeline 处理器
39                    .childHandler(new ChannelInitializer<SocketChannel>() {
40                        @Override
41                        public void initChannel(SocketChannel ch) throws Exception {
42                            // 通过channel获取管道对象
43                            ChannelPipeline pipeline = ch.pipeline();
44                            // 通过管道挂在handler
45                            pipeline.addLast(serverHandler);
46                        }
47                    });
48
49            // 绑定设置的断开（PORT），并阻塞（sync）到绑定完成，绑定完成后就开始监听close信号
50            ChannelFuture f = b.bind().sync();
51
52            // 阻塞到接收client发过来关闭（close）信号
53            f.channel().closeFuture().sync();
54            // 当client发过来close信号后，才会执行这里
55        } finally {
56            // 关闭所有event loop的线程，并关闭内部的线程
57            bossGroup.shutdownGracefully();
58            workerGroup.shutdownGracefully();
59        }
60    }
61
62}

 1import io.netty.buffer.ByteBuf;
 2import io.netty.buffer.Unpooled;
 3import io.netty.channel.ChannelHandler.Sharable;
 4import io.netty.channel.ChannelHandlerContext;
 5import io.netty.channel.ChannelInboundHandlerAdapter;
 6import java.nio.charset.StandardCharsets;
 7
 8/**
 9 * Handler implementation for the echo server.
10 */
11@Sharable
12public class EchoServerHandler extends ChannelInboundHandlerAdapter {
13
14    /**  第一次建立连接调用 */
15    @Override
16    public void channelActive(ChannelHandlerContext ctx) throws Exception {
17        final char[] chars = "hello client, I am Server.".toCharArray();
18        // 建立连接想client发送一句话
19        ctx.pipeline().writeAndFlush(Unpooled.copiedBuffer(chars, StandardCharsets.UTF_8));
20    }
21
22    /** 客户端发送过来数据调用 */
23    @Override
24    public void channelRead(ChannelHandlerContext ctx, Object msg) {
25        final ByteBuf buf = (ByteBuf) msg;
26        String recv = "接收：" + buf.toString(StandardCharsets.UTF_8);
27        if(recv.indexOf("exit") != -1){
28            ctx.close();
29            return;
30        }
31        System.out.print(recv);
32        // 将数据传回客户端
33        ctx.write(Unpooled.copiedBuffer(recv.getBytes(StandardCharsets.UTF_8)));
34    }
35
36    /** 读取完毕，完成一次读事件 */
37    @Override
38    public void channelReadComplete(ChannelHandlerContext ctx) {
39        // flush - 冲刷数据，flush一次便会触发一次或多次client的read事件，反之server也是
40        ctx.flush();
41    }
42
43    /** 捕获异常 */
44    @Override
45    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
46        // 引发异常时关闭连接会
47        cause.printStackTrace();
48        ctx.close();
49    }
50
51}

 1import io.netty.bootstrap.Bootstrap;
 2import io.netty.buffer.Unpooled;
 3import io.netty.channel.*;
 4import io.netty.channel.nio.NioEventLoopGroup;
 5import io.netty.channel.socket.SocketChannel;
 6import io.netty.channel.socket.nio.NioSocketChannel;
 7
 8import java.nio.charset.StandardCharsets;
 9import java.util.Scanner;
10
11/**
12 * 启动client，可以想server发送消息。server的console可以查看client发送的消息。
13 */
14public final class EchoClient {
15
16    static final String HOST = System.getProperty("host", "127.0.0.1");
17    static final int PORT = Integer.parseInt(System.getProperty("port", "8007"));
18
19    public static void main(String[] args) throws Exception {
20        // 配置客户端
21        // client工作线程，每一个线程将和一个EventLoop绑定
22        EventLoopGroup group = new NioEventLoopGroup();
23        try {
24            // 客户端引导其
25            Bootstrap b = new Bootstrap();
26            //配置client启动参数
27            b.group(group)
28                    // 配置数据通道，不同的数据通道决定了IO的方式，例如Nio开头的是非阻塞IO、Oio的是阻塞IO
29                    .channel(NioSocketChannel.class)
30                    // 网络传输的参数，例如网络缓存缓存的大小、是否延迟
31                    .option(ChannelOption.TCP_NODELAY, true)
32                    // 入站、出站处理器
33                    // initializer是在入站只执行一次的处理器
34                    // 它用来与server建立连接、并在pipeline上挂载编解码处理器、自定义处理器
35                    .handler(new ChannelInitializer<SocketChannel>() {
36                        @Override
37                        public void initChannel(SocketChannel ch) throws Exception {
38                            ChannelPipeline p = ch.pipeline();
39                            //p.addLast(new LoggingHandler(LogLevel.INFO));
40                            p.addLast(new EchoClientHandler());
41                        }
42                    });
43
44            // 与server建立连接（connect），并阻塞（sync）到连接建立
45            ChannelFuture f = b.connect(HOST, PORT).sync();
46            // 获取建立的连接
47            final Channel clientChannel = f.channel();
48
49            System.out.println("input 'exit' to EXIT.");
50            Scanner sc = new Scanner(System.in);
51            String str;
52            // 读取并向server发送数据
53            while (!(str = sc.next()).startsWith("exit")) {
54                clientChannel.writeAndFlush(Unpooled.copiedBuffer(str.toCharArray(), StandardCharsets.UTF_8));
55            }
56            // 想server发送断开连接的信号
57            clientChannel.close();
58            // 获取关闭连接后的结果（future），并阻塞（sync）到接收到server发过来的断开连接信号
59            clientChannel.closeFuture().sync();
60        } finally {
61            // 关闭所有的EventLoop，并终止内部的线程
62            group.shutdownGracefully();
63        }
64    }
65}

 1import io.netty.buffer.ByteBuf;
 2import io.netty.buffer.Unpooled;
 3import io.netty.channel.ChannelHandlerContext;
 4import io.netty.channel.ChannelInboundHandlerAdapter;
 5
 6import java.nio.charset.StandardCharsets;
 7
 8/**
 9 * client 端 handler
10 */
11public class EchoClientHandler extends ChannelInboundHandlerAdapter {
12
13    private final ByteBuf firstMessage;
14
15    public EchoClientHandler() {
16        firstMessage = Unpooled.copiedBuffer("start data transfer.\n".toCharArray(), StandardCharsets.UTF_8);
17    }
18
19    /**
20     * 建立连接向server发送一个消息
21     */
22    @Override
23    public void channelActive(ChannelHandlerContext ctx) {
24        ctx.writeAndFlush(firstMessage);
25    }
26
27    /**
28     * client的读事件触发会调用此方法
29     */
30    @Override
31    public void channelRead(ChannelHandlerContext ctx, Object msg) {
32        System.out.println("client recv : " + ((ByteBuf) msg).toString(StandardCharsets.UTF_8));
33    }
34
35    /**
36     * 读取完毕向网络发送数据，flush会触发server的read事件
37     */
38    @Override
39    public void channelReadComplete(ChannelHandlerContext ctx) {
40        ctx.flush();
41    }
42
43    @Override
44    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
45        cause.printStackTrace();
46        ctx.close();
47    }
48
49}

1      public class SquareDecoder extends {@link ByteToMessageDecoder} {
2          @Override
3          public void decode(ChannelHandlerContext ctx,  ByteBuf in, List<Object> out)
4                  throws  Exception {
5              out.add(in.readBytes(in.readableBytes()));
6          }
7      }

1       public class StringHandler extends SimpleChannelInboundHandler<String> {
2            @Override
3           protected void channelRead0(ChannelHandlerContext ctx, String message)
4                   throws Exception {
5               System.out.println(message);
6           }
7       }

1    public static class InboundHandler extends ChannelInboundHandlerAdapter {
2        @Override
3        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
4            // 处理异常
5            System.err.println(this.getClass().getSimpleName() + " ---- " + cause.getMessage());
6            // 向下一个handler传递异常
7            ctx.fireExceptionCaught(cause);
8        }
9    }

 1            ctx.executor().schedule(()->{
 2                channelFuture.addListener(new ChannelFutureListener() {
 3                    @Override
 4                    public void operationComplete(ChannelFuture future) throws Exception {
 5                        System.out.println("writeAndFlush done.");
 6                        if(future.isSuccess()){
 7                            System.out.println("send success.");
 8                        }else{
 9                       		// 处理异常
10                            System.out.println("send fail!");
11                        }
12                    }
13                });
14            }, 0, TimeUnit.SECONDS);

 1    private void invokeExceptionCaught(final Throwable cause) {
 2        if (invokeHandler()) {
 3            try {
 4                //
 5                handler().exceptionCaught(this, cause);
 6            } catch (Throwable error) {
 7                if (logger.isDebugEnabled()) {
 8                    logger.debug(
 9                        "An exception {}" +
10                        "was thrown by a user handler's exceptionCaught() " +
11                        "method while handling the following exception:",
12                        ThrowableUtil.stackTraceToString(error), cause);
13                } else if (logger.isWarnEnabled()) {
14                    logger.warn(
15                        "An exception '{}' [enable DEBUG level for full stacktrace] " +
16                        "was thrown by a user handler's exceptionCaught() " +
17                        "method while handling the following exception:", error, cause);
18                }
19            }
20        } else {
21            fireExceptionCaught(cause);
22        }
23    }

 1    private void invokeChannelActive() {
 2        if (invokeHandler()) {
 3            try {
 4                ((ChannelInboundHandler) handler()).channelInactive(this);
 5            } catch (Throwable t) {
 6            	// 处理异常
 7                invokeExceptionCaught(t);
 8            }
 9        } else {
10            fireChannelInactive();
11        }
12    }
13
14    private void invokeChannelRead(Object msg) {
15        if (invokeHandler()) {
16            try {
17                ((ChannelInboundHandler) handler()).channelRead(this, msg);
18            } catch (Throwable t) {
19            	// 处理异常
20                invokeExceptionCaught(t);
21            }
22        } else {
23            fireChannelRead(msg);
24        }
25    }

 1   // A special catch-all handler that handles both bytes and messages.
 2    final class TailContext extends AbstractChannelHandlerContext implements ChannelInboundHandler {
 3
 4        TailContext(DefaultChannelPipeline pipeline) {
 5            super(pipeline, null, TAIL_NAME, TailContext.class);
 6            setAddComplete();
 7        }
 8
 9        @Override
10        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
11            onUnhandledInboundUserEventTriggered(evt);
12        }
13
14        @Override
15        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
16            // 处理最后的异常，方法如下
17            onUnhandledInboundException(cause);
18        }
19        // 省略...
20    }
21
22    /**
23     * 处理最后的异常
24     * Called once a {@link Throwable} hit the end of the {@link ChannelPipeline} without been handled by the user
25     * in {@link ChannelHandler#exceptionCaught(ChannelHandlerContext, Throwable)}.
26     */
27    protected void onUnhandledInboundException(Throwable cause) {
28        try {
29            logger.warn(
30                    "An exceptionCaught() event was fired, and it reached at the tail of the pipeline. " +
31                            "It usually means the last handler in the pipeline did not handle the exception.",
32                    cause);
33        } finally {
34            ReferenceCountUtil.release(cause);
35        }
36    }
37
38    final class HeadContext extends AbstractChannelHandlerContext
39            implements ChannelOutboundHandler, ChannelInboundHandler {
40
41        private final Unsafe unsafe;
42
43        HeadContext(DefaultChannelPipeline pipeline) {
44            super(pipeline, null, HEAD_NAME, HeadContext.class);
45            unsafe = pipeline.channel().unsafe();
46            setAddComplete();
47        }
48
49        @Override
50        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
51            ctx.fireExceptionCaught(cause);
52        }
53        // 省略...
54    }

1public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit)
2public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit)
3public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)
4public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit)

 1    /**
 2     * 核心方法，所有暴露的schedule方法最后都调用此方法
 3     * @param task 可执行的对象（实现Runnable）且具有返回值（实现Future）
 4     * @return task执行结果，可获取执行结果（success/failure），可以监听状态
 5     */
 6    private <V> ScheduledFuture<V> schedule(final ScheduledFutureTask<V> task) {
 7        // 判断在哪里创建（schedule）定时任务
 8        if (inEventLoop()) {
 9            // 在EventLoop线程中
10            // 如果是在EventLoop（handler执行上下文）中创建的定时任务，就放到任务执行队列中
11            scheduleFromEventLoop(task);
12        } else {
13            // 在外部线程 e.g. main线程调用schedule创建定时任务
14
15            // 截止时间 （当前 - ScheduleFutureTask创建时间）
16            final long deadlineNanos = task.deadlineNanos();
17            // 任务提交前回调判断，true 立即执行任务，false 延迟执行
18            if (beforeScheduledTaskSubmitted(deadlineNanos)) {
19                // execute表示调用执行（立即run），是个未实现的方法，未来由子类实现，由父类调用
20                execute(task);
21            } else {
22                // 在不重写父类lazyExecute时默认仍然使用当前EventLoop线程执行
23                lazyExecute(task);
24                // 任务提交之后回调，方法返回true唤醒当前EventLoop线程，false不唤醒
25                if (afterScheduledTaskSubmitted(deadlineNanos)) {
26                    // 官方解释是为了避免线程竞争
27                    // WAKEUP_TASK这个任务run方法没有执行的语句，使线程保持活跃（等待->可执行，持有CPU资源），避免线程竞争CPU开销
28                    execute(WAKEUP_TASK);
29                }
30            }
31        }
32        return task;
33    }

1    // 在EventLoop中创建的定时任务放在taskQueue, 方便EventGroup调度
2	final void scheduleFromEventLoop(final ScheduledFutureTask<?> task) {
3        // nextTaskId a long and so there is no chance it will overflow back to 0
4        scheduledTaskQueue().add(task.setId(++nextTaskId));
5    }

1private void doStartThread();
2// 执行一个任务
3public void execute(Runnable task);
4// 关闭
5public Future<?> shutdownGracefully(long quietPeriod, long timeout, TimeUnit unit);

1    /**
2     * The NIO {@link Selector}.
3     */
4    private Selector selector;
5    private Selector unwrappedSelector;
6    private SelectedSelectionKeySet selectedKeys;