Java的NIO及Netty

目录:

  1. Java NIO
  2. Netty介绍
  3. Netty在Dubbo中的应用

一、Java NIO

NIO(New IO),和Linux提到的NIO(Non-Blocking IO)并不是一个概念。它主要是利用了IO多路复用的机制。

关于IO模型的知识,之前已经在另外一篇文章中写了: 并发编程之I/O多路复用及几种I/O模式 。通过IO多路复用可以实现Socket的非阻塞操作。后续的Netty也是基于IO多路复用模型实现的。

而IO多路复用机制就是依赖于事件多路分解机制实现的。我们都知道一个完整I/O操作主要有两步:

1、等待数据准备好(写到内核空间);

2、从内科空间拷贝到用户空间;

那么整个NIO主要有几个事件:读就绪、写就绪、有新请求。

事件分发就是在合适的时机,将准备好的读写事件分发给已经注册的事件处理器上。

多路复用就是利用一个Selector不段地去选择一个就绪的事件,并执行。在整个运行期间,选择哪个就绪事件这个函数是阻塞的。

目前比较著名的事件分发模式是Reactor和Proactor。Java的NIO就是基于Reactor模型。

在说Reactor之前,说下NIO的基本概念和核心。

NIO最重要的三大部分:Buffer,Channel,Selector。

Buffer:

在NIO编程中,数据的读写全部都是在Buffer中进行的,它是内存中的一块儿缓冲区,被Java封装成了一个Bugger对象,内部是个数组,属性包括容量,上限,标记,位置等等。配合Channel实现了面向缓冲区的双向处理方式。相比于流处理,因为它是面向整个缓冲区的,块处理更能反映底层操作系统的真实情况,且其中的直接缓冲区还可以减少一次拷贝。此外,Buffer本身提供了非常丰富的方法,如清楚,标记,压缩等等。

Channel:

从单词可以看到,这是一个通道,网络数据的读写都要通过Channel,它是一个双向的通道,两个方向都可以读写,且是全双工的,两个方向可以同时读写。

Java中的Channel对象主要分为两大类,一是用于网络数据操作的SelectableChannel和文件操作的FileChannel。

Selector:

NIO操作是基于IO多路复用的,因此Selector是很重要的一个组件。一个Selector会不断地轮询已经注册的Channel事件,当某个Channel有已经就绪的IO事件,就会进行事件分发。从而实现非阻塞IO。

看下一个基本NIO操作的时序图(来自Netty权威指南):

接下来说下Reactor。

Reactor的基本思想就是不断地监听事件,并将准备好的事件分发给对应的事件处理器。它主要由两个部分组成:Reactor和Handlers。

Reactor:分发合适的handler;

Handlers:执行非阻塞操作。

Reactor主要分3种:单线程Reactor,多线程Reactor以及Worker线程Reactor。

1、单线程

Reactor对象通过select不断监听事件,收到事件后会进行dispatch分发,当有请求过来会分给acceptor并创建handler做后续处理;当不是新请求,就直接分给对应的Hander进行处理。

目前Redis采用的是该模式。该模式的缺点就是不能充分利用多核CPU,因为它是单线程的。

Redis服务器中有两类事件,文件事件和时间事件。

  • 文件事件(file event):Redis客户端通过socket与Redis服务器连接,而文件事件就是服务器对套接字操作的抽象。例如,客户端发了一个GET命令请求,对于Redis服务器来说就是一个文件事件。Redis的协议本身发送的是文本。
  • 时间事件(time event):服务器定时或周期性执行的事件。例如,定期执行RDB持久化。

不过要补充一点,Redis4.0开始也引入了多线程。主要是为了解决一些耗时较长的一些操作,从而提高性能。比如删除大key啊。涉及到的命令有Unlink,FlushAll Async,FlushDB Async.从名字也可以看到,这些命令在执行时,会开启一个异步线程去做处理,而不会长时间阻塞到当前线程。

此外Nginx采用的是多进程+Reactor单线程的方式来处理的,它自己实现了一个Master进程+多个Master进程,从而实现并发。

很多网上的都说Nginx是多Reactor多线程的模式,我也不知道从哪里得来的,Nginx的多进程和多Reactor不是一个概念。因为主进程不会accept连接,而是由子进程的reactor来accept连接,通过锁来控制一次只有一个子进程accept,子进程accept成功后就放到自己的reactor进行处理。Nginx的Master是为了管理这些Worker的。它只是将单Reactor单线程,fork了多个进程而已。

2、多线程

多线程版本和单线程的区别是将实际上做处理业务的非IO操作的业务,如编解码,计算等,放到其他线程处理,这样能充分发挥多核的作用。

虽然多线程版本能发挥多核的作用,但Socket操作不是线程安全的,比如读Socket,因此一般情况下单此按成就可满足了,如果想要多线程,可考虑另一种模式:Proactor

3、多Reactor多线程

用负载均衡的思想去匹配CPU和I/O的速率,

更详细的可见: Scalable IO in Java

二、Netty介绍

由于Java的原生NIO并不友好,需要开发者做很多额外工作,且容易出bug,因此应运而生了一个Netty,它完美解决了Java原生NIO的问题。

我就直接把权威指南关于Netty的优势粘贴过来把。

Netty可以配置使用上述三种Reactor模型中的任何一种。

Netty使用的是Reactor模型,该模型底层是基于IO多路复用的epoll机制,而该机制实现了同步非阻塞的。但是Netty框架实现了异步IO,说异步主要是针对用户来说的,它的异步是通过框架层面实现的,而不是底层IO的读写操作本身。因为Netty中所有的I/O操作都是异步的,所有的调用都会立即返回一个ChannelFuture结果,该结果不一定成功,调用者可以通过该对象获取状态。比如IO操作刚开始时,会新建一个ChannelFuture对象,当完成时,对象状态是已完成。ChannelFuture继承自并发包的Future,该类在线程池的章节有过介绍。

所以,这里请注意,Netty所说的异步和底层采用的epoll的同步非阻塞并不冲突。

常见操作方法:

  • isDone  判断当前操作是否完成。
  • isSuccess 判断已完成的当前操作是否成功。
  • getCause  获取已完成的当前操作失败原因。
  • isCancelled 判断已完成的当前操作是否取消。
  • addListener 注册监听器,当操作已完成(isDone 方法返回true),将会通知指定的监听器。

下面是一张Netty的业务架构图(来自网络):

看到上面的结构,也可以看出它就是多Reactor多线程模型的变种。当客户端请求到达时,Boss Group主要用来accept请求,并进行请求分发(注册Channel到WorkGroup的NIOEventGroup);WorkGroup用来进行I/O事件的处理,它类似一个线程池。

Boss Group和WorkGroup本质都是NioEventLoopGroup,它内部维护了一个线程池,由多个EventLoop组成。而EventLoop除了IO处理外,还负责定时任务以及系统的Task。

Netty解决了原生NIO导致的epoll bug(出现空查询,死循环)。基本思想时会对空查询进行计数,如果超过了N次,即产生了epollBug,Netty会重新建立Selector。

public final class NioEventLoop extends SingleThreadEventLoop {
    private static final InternalLogger logger = InternalLoggerFactory.getInstance(NioEventLoop.class);
    private static final int CLEANUP_INTERVAL = 256;
    private static final boolean DISABLE_KEYSET_OPTIMIZATION = SystemPropertyUtil.getBoolean("io.netty.noKeySetOptimization", false);
    private static final int MIN_PREMATURE_SELECTOR_RETURNS = 3;
    private static final int SELECTOR_AUTO_REBUILD_THRESHOLD;
    private final IntSupplier selectNowSupplier = new IntSupplier() {
        public int get() throws Exception {
            return NioEventLoop.this.selectNow();
        }
    };
    private Selector selector;    //Selector选择器
    private Selector unwrappedSelector;
    private SelectedSelectionKeySet selectedKeys;    //key
    private final SelectorProvider provider;
    private final AtomicBoolean wakenUp = new AtomicBoolean();
    private final SelectStrategy selectStrategy;
    private volatile int ioRatio = 50;
    private int cancelledKeys;
    private boolean needsToSelectAgain;

 
 }

好,到目前为止,上述所述内容都属于通信调度层,有Reactor,Channel,NIOEventLoop等等,负责请求的建立,调度,该层将数据写到内存缓冲区。此后会将后续的读,写等事件触发到PipeLine,后续操作由PipeLine完成。

一个Channel对应一个ChannelPipeLine,而ChannelPipeLine内部又维护了一个双向链表,存储了一系列的Handler。ChannelPipe的作用就是保证有序得执行入站和出站事件,可以选择哪些事件进行拦截或处理。在一系列Handler处理过程中,在任何一个节点都可能拦截或结束。

Channel在上面也提到过,它是进行网络数据读写的I/O接口,Netty并没有直接使用NIO的Channel,而是自己又实现了Channel,因为它要自实现一些具体功能,比如关联pipeline等,即其具备的功能更全。最重要的两个子类:NioSocketChannel和NioServerSocketChannel。

看一下AbstractChannel的属性,有id,有pipeline,有eventloop(选择器),还有Unsafe。Unsafe类是真正操作IO操作的对象。

public abstract class AbstractChannel extends DefaultAttributeMap implements Channel {
  
    private final Channel parent;
    private final ChannelId id;
    private final Unsafe unsafe;
    private final DefaultChannelPipeline pipeline;
    private final VoidChannelPromise unsafeVoidPromise = new VoidChannelPromise(this, false);
    private final AbstractChannel.CloseFuture closeFuture = new AbstractChannel.CloseFuture(this);
    private volatile SocketAddress localAddress;
    private volatile SocketAddress remoteAddress;
    private volatile EventLoop eventLoop;
    private volatile boolean registered;
    private boolean closeInitiated;
    private boolean strValActive;
    private String strVal;

    protected AbstractChannel(Channel parent) {
        this.parent = parent;
        this.id = this.newId();
        this.unsafe = this.newUnsafe();
        this.pipeline = this.newChannelPipeline();
    }

}

每一个Channel都具备的方法,从方法名都可以看出其含义来:

 public boolean isRegistered() {
        return this.registered;
    }

    
    public ChannelFuture bind(SocketAddress localAddress) {
        return this.pipeline.bind(localAddress);
    }

    public ChannelFuture connect(SocketAddress remoteAddress) {
        return this.pipeline.connect(remoteAddress);
    }

    public ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress) {
        return this.pipeline.connect(remoteAddress, localAddress);
    }

    public ChannelFuture disconnect() {
        return this.pipeline.disconnect();
    }

    public ChannelFuture close() {
        return this.pipeline.close();
    }

    public ChannelFuture deregister() {
        return this.pipeline.deregister();
    }

    public Channel flush() {
        this.pipeline.flush();
        return this;
    }

    public ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {
        return this.pipeline.bind(localAddress, promise);
    }

    public ChannelFuture connect(SocketAddress remoteAddress, ChannelPromise promise) {
        return this.pipeline.connect(remoteAddress, promise);
    }

    public ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) {
        return this.pipeline.connect(remoteAddress, localAddress, promise);
    }

    public ChannelFuture disconnect(ChannelPromise promise) {
        return this.pipeline.disconnect(promise);
    }

    public ChannelFuture close(ChannelPromise promise) {
        return this.pipeline.close(promise);
    }

    public ChannelFuture deregister(ChannelPromise promise) {
        return this.pipeline.deregister(promise);
    }

其实在Channel中一个重要的问题就是并发引起的线程安全,比如注册Channel到EventLoop时,不是当前线程发起的,而是其他线程发起的,那么就可能引起问题。针对这个问题,Nety会将其封装成一个Task放到消息队列中,由IO线程完成。比如注册代码:

  public final void register(EventLoop eventLoop, final ChannelPromise promise) {
            if (eventLoop == null) {
                throw new NullPointerException("eventLoop");
            } else if (AbstractChannel.this.isRegistered()) {
                promise.setFailure(new IllegalStateException("registered to an event loop already"));
            } else if (!AbstractChannel.this.isCompatible(eventLoop)) {
                promise.setFailure(new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName()));
            } else {
                AbstractChannel.this.eventLoop = eventLoop;
                //如果是当前线程,直接注册
                if (eventLoop.inEventLoop()) {
                    this.register0(promise);
                   //如果不是当前线程,封装成Task
                } else {
                    try {
                        eventLoop.execute(new Runnable() {
                            public void run() {
                                AbstractUnsafe.this.register0(promise);
                            }
                        });
                    } catch (Throwable var4) {
                        AbstractChannel.logger.warn("Force-closing a channel whose registration task was not accepted by an event loop: {}", AbstractChannel.this, var4);
                        this.closeForcibly();
                        AbstractChannel.this.closeFuture.setClosed();
                        this.safeSetFailure(promise, var4);
                    }
                }

            }
        }

ChannelHandler用来处理或者拦截IO操作,比如编解码或者其他自定义的操作。看例子:


    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();
        pipeline.addLast("framer", new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
        pipeline.addLast("decoder", new StringDecoder(CharsetUtil.UTF_8)); //解码,将ByteBuf二进制数据解码成字符串并向后传播
        pipeline.addLast("encoder", new StringEncoder(CharsetUtil.UTF_8));  //内置编码
        pipeline.addLast("handler", new NettyServerHandler());     //自定义
    }

//自定义的Handler

@Slf4j
public class NettyServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext channelHandlerContext, Object s) throws Exception {
        log.info("netty server 成功接收来自:{}的消息:{}",channelHandlerContext.channel().remoteAddress(),s);
        channelHandlerContext.writeAndFlush("nettyServer已经成功接收到您的消息!\n");
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("客户端[ip:" + ctx.channel().remoteAddress() + "]连接");
        ctx.writeAndFlush("CONNECT SUCCESS\n");
        super.channelActive(ctx);
    }

    /**
     * 异常
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        try {
            log.error("Netty服务出现异常", cause);
            // 当出现异常就关闭连接
            ctx.close().sync();
        } catch (Exception e) {
            log.error("Netty服务出现异常", e);
        }

    }

}

ChannelHandler主要分为入站和出站。

  • ChannelInboundHandler 用于处理入站 I/O 事件。
  • ChannelOutboundHandler 用于处理出站 I/O 操作。

对应的Adapter:

  • ChannelInboundHandlerAdapter 用于处理入站 I/O 事件。
  • ChannelOutboundHandlerAdapter 用于处理出站 I/O 操作。
  • ChannelDuplexHandler 用于处理入站和出站事件。

这里我们说下ChannelHandler的安全性。假如一个Handler被注册到不同的线程中,如果多个线程都要执行该Handler势必会出现线程安全额问题。针对这个问题,Netty的做法是检查这个Handler是否Sharable注解的,只有被Sharable注解的才可能加到不同的Channel中。

此外,对于读写操作。Netty4只有一个NioEventLoop线程来处理这个操作,业务耗时ChannelHandler被I/O线程串行执行,所以执行效率低。Netty3在消息发送线程模型上,充分利用业务线程的并行编码和ChanelHandler的优势,在一个周期T内可以处理N条业务消息。

下面时ChannelPipeLine的处理流程:

其中入站事件一般由IO线程触发,比如TCP链路建立,读事件,异常通知等。出站事件一般由用户主动发起,发送消息等。

我们建立一个Netty Server的基本代码:

@Component
@Slf4j
public class YPNettyServer {

    private EventLoopGroup boss = new NioEventLoopGroup();

    private EventLoopGroup work = new NioEventLoopGroup();

    /**
     * 最好不要用注解启动,否则会阻塞,影响springboot server
     *
     * @throws InterruptedException
     */
//    @PostConstruct
    public void start() throws InterruptedException {
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(boss,work)
                .channel(NioServerSocketChannel.class)
                .localAddress(new InetSocketAddress(6666))
                .option(ChannelOption.SO_BACKLOG,1024)
                .childOption(ChannelOption.SO_KEEPALIVE,true)
                .childOption(ChannelOption.TCP_NODELAY,true)
                .childHandler(new NettyServerInitializer());

        try {
            ChannelFuture channelFuture = bootstrap.bind(6666).sync();
            log.info("nettyServer开始监听接口:6666");
            channelFuture.channel().closeFuture().sync();
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            boss.shutdownGracefully();
            work.shutdownGracefully();
        }
    }
}

NIOEventLoopGroup

三、Netty在Dubbo中的实践

Dubbo官方文档介绍得非常详细: Dubbo服务调用过程

在服务端,我们通常使用Service注解来暴露dubbo服务,在springboot启动时,会创建一个ServiceBean。

该类继承自InitializingBean,ApplicationListener等。

该类中的onApplicationEvent会在启动过程中被调用,会调用export方法,该方法会调用DubboProtocol的openServer,接着调用createServer,该方法中会利用Dubbo的SPI机制,调用NettyServer.关于DubboSPI的介绍文章: Dubbo源码之SPI机制

  private ExchangeServer createServer(URL url) {
        url = url.addParameterIfAbsent("channel.readonly.sent", Boolean.TRUE.toString());
        url = url.addParameterIfAbsent("heartbeat", String.valueOf(60000));
        String str = url.getParameter("server", "netty");
        if (str != null && str.length() > 0 && !ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str)) {
            throw new RpcException("Unsupported server type: " + str + ", url: " + url);
        } else {
            url = url.addParameter("codec", "dubbo");

            ExchangeServer server;
            try {
                server = Exchangers.bind(url, this.requestHandler);
            } catch (RemotingException var5) {
                throw new RpcException("Fail to start server(url: " + url + ") " + var5.getMessage(), var5);
            }

            str = url.getParameter("client");
            if (str != null && str.length() > 0) {
                Set<String> supportedTypes = ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions();
                if (!supportedTypes.contains(str)) {
                    throw new RpcException("Unsupported client type: " + str);
                }
            }

            return server;
        }
    }

最终会调用NettyServer的doOpen.

protected void doOpen() throws Throwable {
        this.bootstrap = new ServerBootstrap();
        this.bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory("NettyServerBoss", true));
        this.workerGroup = new NioEventLoopGroup(this.getUrl().getPositiveParameter("iothreads", Constants.DEFAULT_IO_THREADS), new DefaultThreadFactory("NettyServerWorker", true));
        final NettyServerHandler nettyServerHandler = new NettyServerHandler(this.getUrl(), this);
        this.channels = nettyServerHandler.getChannels();
        ((ServerBootstrap)this.bootstrap.group(this.bossGroup, this.workerGroup).channel(NioServerSocketChannel.class)).childOption(ChannelOption.TCP_NODELAY, Boolean.TRUE).childOption(ChannelOption.SO_REUSEADDR, Boolean.TRUE).childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT).childHandler(new ChannelInitializer<NioSocketChannel>() {
            protected void initChannel(NioSocketChannel ch) throws Exception {
                NettyCodecAdapter adapter = new NettyCodecAdapter(NettyServer.this.getCodec(), NettyServer.this.getUrl(), NettyServer.this);
                ch.pipeline().addLast("decoder", adapter.getDecoder()).addLast("encoder", adapter.getEncoder()).addLast("handler", nettyServerHandler);
            }
        });
        ChannelFuture channelFuture = this.bootstrap.bind(this.getBindAddress());
        channelFuture.syncUninterruptibly();
        this.channel = channelFuture.channel();
    }

这个方法和我在上面建立的简单的Netty服务器区别不大,还是很简单的。

Dubbo采用的是一个主Reactor,多个从Reactor多线程模型。

当服务端收到客户端的请求时,首先Dubbo首先解码数据,解码过程不说了,可以参考官方文档,解码后会封装一个Request对象,然后将请求发送至DisPatcher分发调度器,然后分发器会将请求分发至线程池,线程池负责执行具体任务。

Netty4选择的默认分发策略是all,即将所有消息都分发到业务线程池中执行。

 public NettyServer(URL url, ChannelHandler handler) throws RemotingException {
        super(url, ChannelHandlers.wrap(handler, ExecutorUtil.setThreadName(url, "DubboServerHandler")));
    }


 protected ChannelHandler wrapInternal(ChannelHandler handler, URL url) {
        return new MultiMessageHandler(new HeartbeatHandler(((Dispatcher)ExtensionLoader.getExtensionLoader(Dispatcher.class).getAdaptiveExtension()).dispatch(handler, url)));
    }

除了上面所说的Handler,还有一个DecodeHandler.

在上面的createServer代码,还有一句:

server = Exchangers.bind(url, this.requestHandler);
 public static Exchanger getExchanger(String type) {
        return (Exchanger)ExtensionLoader.getExtensionLoader(Exchanger.class).getExtension(type);
    }

利用Dubbo SPI机制,加载HeaderExchanger对象。

  public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
        return new HeaderExchangeServer(Transporters.bind(url, new ChannelHandler[]{new DecodeHandler(new HeaderExchangeHandler(handler))}));
    }

如果只暴露一个Dubbo服务端口,那么一只有一个NettyHandler。此外设置了用来处理编解码、连接建立和消息读写等的Handler,即MultiMessageHandler,HeartbeatHandler以及分发器实现类,比如AllChannelHandler,还有DecodecHandler。

MultiMessageHandler->HeartbeatHandler->AllChannelHandler->DecodecHandler-> HeaderExchangeHandler。

处理流程图(来自网络):

具体流程:

当解码器将请求封装成Request对象后,NettyServerHandler的channelRead会收到信息;

随后会调用ChannelHandler的recevied方法,依次是:

- MultiMessageHandler.received;主要处理批量消息;

- HeartbeatHandler.received;心跳消息;

- AllChannelHandler.received;Dispatcher分发器

-任务被派发之后,就会具体调用服务了,这里旧不赘述了。

看下AllChannelHandler的received方法:

 public void received(Channel channel, Object message) throws RemotingException {
        //线程池
        ExecutorService cexecutor = this.getExecutorService();

        try {
            cexecutor.execute(new ChannelEventRunnable(channel, this.handler, ChannelState.RECEIVED, message));
        } catch (Throwable var8) {
            if (message instanceof Request && var8 instanceof RejectedExecutionException) {
                Request request = (Request)message;
                if (request.isTwoWay()) {
                    String msg = "Server side(" + this.url.getIp() + "," + this.url.getPort() + ") threadpool is exhausted ,detail msg:" + var8.getMessage();
                    Response response = new Response(request.getId(), request.getVersion());
                    response.setStatus((byte)100);
                    response.setErrorMessage(msg);
                    channel.send(response);
                    return;
                }
            }

            throw new ExecutionException(message, channel, this.getClass() + " error when process received event .", var8);
        }
    }

其线程池实例也是通过Dubbo SPI机制加载:

 public WrappedChannelHandler(ChannelHandler handler, URL url) {
        this.handler = handler;
        this.url = url;
        this.executor = (ExecutorService)((ThreadPool)ExtensionLoader.getExtensionLoader(ThreadPool.class).getAdaptiveExtension()).getExecutor(url);
        String componentKey = Constants.EXECUTOR_SERVICE_COMPONENT_KEY;
        if ("consumer".equalsIgnoreCase(url.getParameter("side"))) {
            componentKey = "consumer";
        }

        DataStore dataStore = (DataStore)ExtensionLoader.getExtensionLoader(DataStore.class).getDefaultExtension();
        dataStore.put(componentKey, Integer.toString(url.getPort()), this.executor);
    }

Dubbo很好地融合了Netty,选择了多Reactor多线程的模型,充分提高了并发性能。看了里面的代码,觉得Dubbo真的很不错。接下来要好好学习一下Dubbo。

参考文章:

--------EOF---------
本文微信分享/扫码阅读