Netty组件

EventLoop

事件循环对象

EventLoop本质是一个单线程执行器(同时维护了一个Selector，里面有run方法处理Channel上源源不断的io事件

它的继承关系比较复杂

• 一条线是继承自 j.u.c.ScheduledExecutorService 因此包含了线程池中所有的方法
• 另一条线是继承自 netty 自己的 OrderedEventExecutor，
  • 提供了 boolean inEventLoop(Thread thread) 方法判断一个线程是否属于此 EventLoop
  • 提供了 parent 方法来看看自己属于哪个 EventLoopGroup

事件循环组

EventLoopGroup 是一组 EventLoop，Channel 一般会调用 EventLoopGroup 的 register 方法来绑定其中一个 EventLoop，后续这个 Channel 上的 io 事件都由此 EventLoop 来处理（保证了 io 事件处理时的线程安全）

• 继承自 netty 自己的 EventExecutorGroup
  • 实现了 Iterable 接口提供遍历 EventLoop 的能力
  • 另有 next 方法获取集合中下一个 EventLoop

示例

 		// 内部创建了两个 EventLoop, 每个 EventLoop 维护一个线程
        EventLoopGroup eventLoopGroup=new NioEventLoopGroup(2);
        EventLoop event = eventLoopGroup.next();//获取EventLoop对象，当遍历到尾节点后又从头开始
        System.out.println(event);
        EventLoop event2 = eventLoopGroup.next();
        System.out.println(event2);
        EventLoop event3 = eventLoopGroup.next();
        System.out.println(event3);

输出

io.netty.channel.nio.NioEventLoop@282ba1e
io.netty.channel.nio.NioEventLoop@13b6d03
io.netty.channel.nio.NioEventLoop@282ba1e

处理普通与定时任务

普通任务

 EventLoop event = eventLoopGroup.next();//获取EventLoop对象，当遍历到尾节点后又从头开始
        //执行普通任务
        event.submit(()->{
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            System.out.println("run task ...");
        });
        event.shutdownGracefully();

定时任务

        event.scheduleAtFixedRate(()->{
            System.out.println("run schedule task ...");
        },0,1,TimeUnit.SECONDS);

        System.out.println("main ...");

💡关闭 EventLoopGroup

优雅关闭 shutdownGracefully 方法。该方法会首先切换 EventLoopGroup 到关闭状态从而拒绝新的任务的加入，然后在任务队列的任务都处理完成后，停止线程的运行。从而确保整体应用是在正常有序的状态下退出的

处理IO任务

服务器端代码

public class Server {
    public static void main(String[] args) {
        new ServerBootstrap().
                group(new NioEventLoopGroup())
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline().addLast(new ChannelInboundHandlerAdapter(){
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                ByteBuf buf = (ByteBuf) msg;
                                System.out.println(buf.toString(StandardCharsets.UTF_8) + "=>"+Thread.currentThread().getName());
                            }
                        });
                    }
                })
                .bind(8888);
    }
}

客户端代码

public class Client {
    public static void main(String[] args) throws InterruptedException, IOException {
        Channel channel = new Bootstrap()
                .group(new NioEventLoopGroup())
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline().addLast(new StringEncoder());
                    }
                }).connect(new InetSocketAddress("localhost", 8888))
                .sync()
                .channel();
        channel.writeAndFlush("haha");
    }
}

Channel会与EventLoop进行绑定

分工与增加自定义EventLoopGroup

Bootstrap的group()方法可以传入两个EventLoopGroup参数，参数1为BossEventGroup,参数2为WorkerEventGroup,单个参数的方法会为Boss与Worker创建相同的EventGroup

  public ServerBootstrap group(EventLoopGroup group) {  //单参方法
        return this.group(group, group);
    }

  public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup) { //双参方法
        super.group(parentGroup);
        if (this.childGroup != null) {
            throw new IllegalStateException("childGroup set already");
        } else {
            this.childGroup = (EventLoopGroup)ObjectUtil.checkNotNull(childGroup, "childGroup");
            return this;
        }
    }

当有的任务需要较长的时间处理时，可以使用非NioEventLoopGroup，避免同一个NioEventLoop中的其他Channel在较长的时间内都无法得到处理

服务器端代码

public class Server {
    public static void main(String[] args) {
        EventLoopGroup group=new DefaultEventLoopGroup();//处理耗时任务，防止阻塞worker thread
        new ServerBootstrap()
                //arg1： BossEventLoopGroup                 arg2:WorkerEventGroup
                .group(new NioEventLoopGroup(1), new NioEventLoopGroup(2))
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                ByteBuf buf = (ByteBuf) msg;
                                System.out.println(Thread.currentThread().getName() + " " + buf.toString(StandardCharsets.UTF_8));
                                ctx.fireChannelRead(msg); //传递msg给下一个handler
                            }
                        }).addLast(group,"handlerA",new ChannelInboundHandlerAdapter() {
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                ByteBuf buf = (ByteBuf) msg;
                                System.out.println(Thread.currentThread().getName() + " " + buf.toString(StandardCharsets.UTF_8));
                            }
                        });
                    }
                })
                .bind(8888);
    }
}

客户端代码

public class Client {
    public static void main(String[] args) throws InterruptedException, IOException {
        Channel channel = new Bootstrap()
                .group(new NioEventLoopGroup())
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline().addLast(new StringEncoder());
                    }
                }).connect(new InetSocketAddress("localhost", 8888))
                .sync()
                .channel();
        channel.writeAndFlush("hello");
    }
}

输出

defaultEventLoopGroup-2-3 1
nioEventLoopGroup-4-1 1
defaultEventLoopGroup-2-3 1
nioEventLoopGroup-4-2 2
defaultEventLoopGroup-2-4 2
nioEventLoopGroup-4-2 2
defaultEventLoopGroup-2-4 2
nioEventLoopGroup-4-1 3
defaultEventLoopGroup-2-5 3

可以看到，nio 工人和 非 nio 工人也分别绑定了 channel

💡 Handler执行中如何换人?

源码:io.netty.channel.AbstractChannelHandlerContext#invokeChannelRead()

static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {
    final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);
    // 获得下一个EventLoop, excutor即为 EventLoopGroup
    EventExecutor executor = next.executor();
    
    // 如果下一个EventLoop 在当前的 EventLoopGroup中
    if (executor.inEventLoop()) {
        // 使用当前 EventLoopGroup 中的 EventLoop 来处理任务
        next.invokeChannelRead(m);
    } else {
        // 否则让另一个 EventLoopGroup 中的 EventLoop 来创建任务并执行
        executor.execute(new Runnable() {
            public void run() {
                next.invokeChannelRead(m);
            }
        });
    }
}

• 如果两个 handler 绑定的是同一个线程，那么就直接调用
• 否则，把要调用的代码封装为一个任务对象，由下一个 handler 的线程来调用

Channel

channel的主要作用

• close():可以关闭channel
• closeFuture():获取一个用来可以用来处理channel关闭的Future对象
  • sync方法的作用是同步等待channel关闭
  • addListener方法是用于异步等待channel关闭
• pipline()方法添加处理器
• wirte()方法将数据写入
• writeAndFlush()方法将数据写入并刷出

ChannelFuture

public class Client {
    public static void main(String[] args) throws InterruptedException, IOException {
        ChannelFuture future = new Bootstrap()
                .group(new NioEventLoopGroup())
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline().addLast(new StringEncoder());
                    }
                }).connect(new InetSocketAddress("localhost", 8888));
                //.sync()  //不使用sync方法则返回未连接成功的channel对象[id: 0x4218fad1]
                future.addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture channelFuture) throws Exception {
                        Channel channel = channelFuture.channel();
                        System.out.println(channel); //返回已连接的channel [id: 0x41bbc250, L:/127.0.0.1:57536 - R:localhost/127.0.0.1:8888]
                        channel.writeAndFlush("async ...");
                    }
                });
    }
}

• 调用connect方法后可以返回一个 ChannelFuture 对象，它的作用是利用 channel() 方法来获取 Channel 对象
• connect 方法是异步的，意味着不等连接建立，方法执行就返回了。因此 channelFuture 对象中不能【立刻】获得到正确的 Channel 对象,需要调用sync方法进行同步阻塞或着调用addListener方法添加异步回调

CloseFuture

@Slf4j
public class ClientClose {
    public static void main(String[] args) throws InterruptedException {
        NioEventLoopGroup group = new NioEventLoopGroup();
        ChannelFuture channelFuture = new Bootstrap()
                .group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
                        socketChannel.pipeline().addLast(new StringEncoder());
                    }
                })
                .connect(new InetSocketAddress("localhost", 8888));
        Channel channel = channelFuture.sync()
                .channel();
        log.info("connect success! " + channel);
        new Thread(() -> {
            Scanner scanner = new Scanner(System.in);
            while (true) {
                String msg = scanner.nextLine();
                if ("q".equals(msg)) {
                    //close channel
                    channel.close(); // async
                    break;
                }
                log.info("send msg:" + msg);
                channel.writeAndFlush(msg);
            }
        }, "myThread").start();
        ChannelFuture closedFuture = channel.closeFuture();
//        closedFuture.sync();
        closedFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture channelFuture) throws Exception {
                log.info("channel closed！....");
                group.shutdownGracefully();
            }
        });
    }
}

• 因为channel.close方法为异步方法，所以对于channel关闭后的操作需要针对性处理
• 调用channel.closeFuture()方法会返回一个ChannelFuture对象
  • 可以通过sync进行阻塞，直到future执行完成
  • 可以通过addListener添加异步回调