宜人贷蜂巢API网关技术解密之Netty使用实践

宜人贷蜂巢团队，由Michael创立于2013年，通过使用互联网科技手段助力金融生态和谐健康发展。自成立起一直致力于多维度数据闭环平台建设。目前团队规模超过百人，涵盖征信、电商、金融、社交、五险一金和保险等用户授信数据的抓取解析业务，辅以先进的数据分析、挖掘和机器学习等技术对用户信用级别、欺诈风险进行预测评定，全面对外输出金融反欺诈、社交图谱、自动化模型定制等服务或产品。

目前宜人贷蜂巢基于用户授权数据实时抓取解析技术，并结合顶尖大数据技术，快速迭代和自主的创新,已形成了强大而领先的聚合和输出能力。

为了适应完成宜人贷蜂巢强大的服务输出能力，蜂巢设计开发了自己的API网关系统，集中实现了鉴权、加解密、路由、限流等功能，使各业务抓取团队关注其核心抓取和分析工作，而API网关系统更专注于安全、流量、路由等问题，从而更好的保障蜂巢服务系统的质量。今天带着大家解密API网关的Netty线程池技术实践细节。

API网关作为宜人贷蜂巢数据开放平台的统一入口，所有的客户端及消费端通过统一的API来使用各类抓取服务。从面向对象设计的角度看，它与外观模式类似，包装各类不同的实现细节，对外表现出统一的调用形式。

本文首先，简要地介绍API网关的项目框架，其次对比BIO和NIO的特点，再引入Netty作为项目的基础框架，然后介绍Netty线程池的原理，最后深入Netty线程池的初始化、ServerBootstrap的初始化与启动及channel与线程池的绑定过程，让读者了解Netty在承载高并发访问的设计路思。

项目框架

图1 - API网关项目框架

图中描绘了API网关系统的处理流程，以及与服务注册发现、日志分析、报警系统、各类爬虫的关系。其中API网关系统接收请求，对请求进行编解码、鉴权、限流、加解密，再基于Eureka服务注册发现模块，将请求发送到有效的服务节点上；网关及抓取系统的日志，会被收集到elk平台中，做业务分析及报警处理。

BIO vs NIO

API网关承载数倍于爬虫的流量，提升服务器的并发处理能力、缩短系统的响应时间，通信模型的选择是至关重要的，是选择BIO，还是NIO？

Streamvs Buffer & 阻塞 vs 非阻塞

BIO是面向流的，io的读写，每次只能处理一个或者多个bytes，如果数据没有读写完成，线程将一直等待于此，而不能暂时跳过io或者等待io读写完成异步通知，线程滞留在io读写上，不能充分利用机器有限的线程资源，造成server的吞吐量较低，见图2。而NIO与此不同，面向Buffer，线程不需要滞留在io读写上，采用操作系统的epoll模式，在io数据准备好了，才由线程来处理，见图3。

图2 – BIO 从流中读取数据

图3 – NIO 从Buffer中读取数据

Selectors

NIO的selector使一个线程可以监控多个channel的读写，多个channel注册到一个selector上，这个selector可以监测到各个channel的数据准备情况，从而使用有限的线程资源处理更多的连接，见图4。所以可以这样说，NIO极大的提升了服务器接受并发请求的能力，而服务器性能还是要取决于业务处理时间和业务线程池模型。

图4 – NIO 单一线程管理多个连接

而BIO采用的是request-per-thread模式，用一个线程负责接收TCP连接请求，并建立链路，然后将请求dispatch给负责业务逻辑处理的线程，见图5。一旦访问量过多，就会造成机器的线程资源紧张，造成请求延迟，甚至服务宕机。

图5 – BIO 一连接一线程

对比JDK NIO与诸多NIO框架后，鉴于Netty优雅的设计、易用的API、优越的性能、安全性支持、API网关使用Netty作为通信模型，实现了基础框架的搭建。

Netty线程池

考虑到API网关的高并发访问需求，线程池设计，见图6。

图6 – API网关线程池设计

Netty的线程池理念有点像ForkJoinPool，不是一个线程大池子并发等待一条任务队列，而是每条线程都有一个任务队列。而且Netty的线程，并不只是简单的阻塞地拉取任务，而是在每个循环中做三件事情：

• 先SelectKeys()处理NIO的事件
• 然后获取本线程的定时任务，放到本线程的任务队列里
• 最后执行其他线程提交给本线程的任务

每个循环里处理NIO事件与其他任务的时间消耗比例，还能通过ioRatio变量来控制，默认是各占50%。可见，Netty的线程根本没有阻塞等待任务的清闲日子，所以也不使用有锁的BlockingQueue来做任务队列了，而是使用无锁的MpscLinkedQueue(Mpsc 是Multiple Producer, Single Consumer的缩写)

NioEventLoopGroup初始化

下面分析下Netty线程池NioEventLoopGroup的设计与实现细节，NioEventLoopGroup的类层次关系见图7

图7 –NioEvenrLoopGroup类层次关系

其创建过程——方法调用，见下图

图8 –NioEvenrLoopGroup创建调用关系

NioEvenrLoopGroup的创建，具体执行过程是执行类MultithreadEventExecutorGroup的构造方法

/**

 * Create a new instance.

 *

 * @param nThreads          the number of threads that will be used by this instance.

 * @param executor          the Executor to use, or {@code null} if the default should be used.

 * @param chooserFactory    the {@link EventExecutorChooserFactory} to use.

 * @param args              arguments which will passed to each {@link #newChild(Executor, Object...)} call

 */

protected MultithreadEventExecutorGroup(int nThreads, Executor executor,

                                        EventExecutorChooserFactory chooserFactory, Object... args) {

    if (nThreads <= 0) {

        throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads));

    }

    if (executor == null) {

        executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());

    }

    children = new EventExecutor[nThreads];

    for (int i = 0; i < nThreads; i ++) {

        boolean success = false;

        try {

            children[i] = newChild(executor, args);

            success = true;

        } catch (Exception e) { 

            throw new IllegalStateException("failed to create a child event loop", e);

        } finally {

            if (!success) {

                for (int j = 0; j < i; j ++) {

                    children[j].shutdownGracefully();

                }

                for (int j = 0; j < i; j ++) {

                    EventExecutor e = children[j];

                    try {

                        while (!e.isTerminated()) {

                            e.awaitTermination(Integer.MAX_VALUE, TimeUnit.SECONDS);

                        }

                    } catch (InterruptedException interrupted) {

                        // Let the caller handle the interruption.

                        Thread.currentThread().interrupt();

                        break;

                    }

                }

            }

        }

    }

    chooser = chooserFactory.newChooser(children);

    final FutureListener<Object> terminationListener = new FutureListener<Object>() {

        @Override

        public void operationComplete(Future<Object> future) throws Exception {

            if (terminatedChildren.incrementAndGet() == children.length) {

                terminationFuture.setSuccess(null);

            }

        }

    };

    for (EventExecutor e: children) {

        e.terminationFuture().addListener(terminationListener);

    }

    Set<EventExecutor> childrenSet = new LinkedHashSet<EventExecutor>(children.length);

    Collections.addAll(childrenSet, children);

    readonlyChildren = Collections.unmodifiableSet(childrenSet);

}

其中，创建细节见下：

• 线程池中的线程数nThreads必须大于0；
• 如果executor为null，创建默认executor，executor用于创建线程（newChild方法使用executor对象）；
• 依次创建线程池中的每一个线程即NioEventLoop，如果其中有一个创建失败，将关闭之前创建的所有线程；
• chooser为线程池选择器，用来选择下一个EventExecutor，可以理解为，用来选择一个线程来执行task；

chooser的创建细节，见下

DefaultEventExecutorChooserFactory根据线程数创建具体的EventExecutorChooser，线程数如果等于2^n，可使用按位与替代取模运算，节省cpu的计算资源，见源码

@SuppressWarnings("unchecked")

@Override

public EventExecutorChooser newChooser(EventExecutor[] executors) {

    if (isPowerOfTwo(executors.length)) {

        return new PowerOfTowEventExecutorChooser(executors);

    } else {

        return new GenericEventExecutorChooser(executors);

    }

} 

    private static final class PowerOfTowEventExecutorChooser implements EventExecutorChooser {

        private final AtomicInteger idx = new AtomicInteger();

        private final EventExecutor[] executors;



        PowerOfTowEventExecutorChooser(EventExecutor[] executors) {

            this.executors = executors;

        }



        @Override

        public EventExecutor next() {

            return executors[idx.getAndIncrement() & executors.length - 1];

        }

    }



    private static final class GenericEventExecutorChooser implements EventExecutorChooser {

        private final AtomicInteger idx = new AtomicInteger();

        private final EventExecutor[] executors;



        GenericEventExecutorChooser(EventExecutor[] executors) {

            this.executors = executors;

        }



        @Override

        public EventExecutor next() {

            return executors[Math.abs(idx.getAndIncrement() % executors.length)];

        }

    }

newChild(executor, args)的创建细节，见下

MultithreadEventExecutorGroup的newChild方法是一个抽象方法，故使用NioEventLoopGroup的newChild方法，即调用NioEventLoop的构造函数

@Override

    protected EventLoop newChild(Executor executor, Object... args) throws Exception {

        return new NioEventLoop(this, executor, (SelectorProvider) args[0],

            ((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2]);

    }

在这里先看下NioEventLoop的类层次关系

NioEventLoop的继承关系比较复杂，在AbstractScheduledEventExecutor 中, Netty 实现了 NioEventLoop 的 schedule 功能, 即我们可以通过调用一个 NioEventLoop 实例的 schedule 方法来运行一些定时任务. 而在 SingleThreadEventLoop 中, 又实现了任务队列的功能, 通过它, 我们可以调用一个NioEventLoop 实例的 execute 方法来向任务队列中添加一个 task, 并由 NioEventLoop 进行调度执行.

通常来说, NioEventLoop 肩负着两种任务, 第一个是作为 IO 线程, 执行与 Channel 相关的 IO 操作, 包括调用 select 等待就绪的 IO 事件、读写数据与数据的处理等; 而第二个任务是作为任务队列, 执行 taskQueue 中的任务, 例如用户调用 eventLoop.schedule 提交的定时任务也是这个线程执行的.

具体的构造过程，见下

创建任务队列tailTasks（内部为有界的LinkedBlockingQueue）

创建线程的任务队列taskQueue（内部为有界的LinkedBlockingQueue），以及任务过多防止系统宕机的拒绝策略rejectedHandler

其中tailTasks和taskQueue均是任务队列，而优先级不同，taskQueue的优先级高于tailTasks，定时任务的优先级高于taskQueue。

ServerBootstrap初始化及启动

了解了Netty线程池NioEvenrLoopGroup的创建过程后，下面看下API网关服务ServerBootstrap的是如何使用线程池引入服务中，为高并发访问服务的。

API网关ServerBootstrap初始化及启动代码，见下

serverBootstrap = new ServerBootstrap();

bossGroup = new NioEventLoopGroup(config.getBossGroupThreads());

workerGroup = new NioEventLoopGroup(config.getWorkerGroupThreads());



serverBootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)

        .option(ChannelOption.TCP_NODELAY, config.isTcpNoDelay())

        .option(ChannelOption.SO_BACKLOG, config.getBacklogSize())

        .option(ChannelOption.SO_KEEPALIVE, config.isSoKeepAlive())

        // Memory pooled

        .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)

        .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)

        .childHandler(channelInitializer);

 

ChannelFuture future = serverBootstrap.bind(config.getPort()).sync();

log.info("API-gateway started on port: {}", config.getPort());

future.channel().closeFuture().sync();

API网关系统使用netty自带的线程池，共有三组线程池，分别为bossGroup、workerGroup和executorGroup（使用在channelInitializer中，本文暂不作介绍）。其中，bossGroup用于接收客户端的TCP连接，workerGroup用于处理I/O、执行系统task和定时任务，executorGroup用于处理网关业务加解密、限流、路由，及将请求转发给后端的抓取服务等业务操作。

Channel与线程池的绑定

ServerBootstrap初始化后，通过调用bind(port)方法启动Server，bind的调用链如下

AbstractBootstrap.bind ->AbstractBootstrap.doBind -> AbstractBootstrap.initAndRegister

其中，ChannelFuture regFuture = config().group().register(channel);中的group()方法返回bossGroup，而channel在serverBootstrap的初始化过程指定channel为NioServerSocketChannel.class，至此将NioServerSocketChannel与bossGroup绑定到一起，bossGroup负责客户端连接的建立。那么NioSocketChannel是如何与workerGroup绑定到一起的？

调用链AbstractBootstrap.initAndRegister -> AbstractBootstrap. init-> ServerBootstrap.init ->ServerBootstrapAcceptor.ServerBootstrapAcceptor ->ServerBootstrapAcceptor.channelRead

public void channelRead(ChannelHandlerContext ctx, Object msg) {

    final Channel child = (Channel) msg;

    child.pipeline().addLast(childHandler);

    for (Entry<ChannelOption<?>, Object> e: childOptions) {

        try {

            if (!child.config().setOption((ChannelOption<Object>) e.getKey(), e.getValue())) {

                logger.warn("Unknown channel option: " + e);

            }

        } catch (Throwable t) {

            logger.warn("Failed to set a channel option: " + child, t);

        }

    }

    for (Entry<AttributeKey<?>, Object> e: childAttrs) {

        child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());

    }

    try {

        childGroup.register(child).addListener(new ChannelFutureListener() {

            @Override

            public void operationComplete(ChannelFuture future) throws Exception {

                if (!future.isSuccess()) {

                    forceClose(child, future.cause());

                }

            }

        });

    } catch (Throwable t) {

        forceClose(child, t);

    }

}

其中，childGroup.register(child)就是将NioSocketChannel与workderGroup绑定到一起，那又是什么触发了ServerBootstrapAcceptor的channelRead方法？

其实当一个 client 连接到 server 时, Java 底层的 NIO ServerSocketChannel 会有一个SelectionKey.OP_ACCEPT 就绪事件, 接着就会调用到 NioServerSocketChannel.doReadMessages方法

@Override

protected int doReadMessages(List<Object> buf) throws Exception {

    SocketChannel ch = javaChannel().accept();

    try {

        if (ch != null) {

            buf.add(new NioSocketChannel(this, ch));

            return 1;

        }

    } catch (Throwable t) {

        …

    }

    return 0;

}

javaChannel().accept() 会获取到客户端新连接的SocketChannel,实例化为一个 NioSocketChannel, 并且传入 NioServerSocketChannel 对象(即 this), 由此可知, 我们创建的这个NioSocketChannel 的父 Channel 就是 NioServerSocketChannel 实例 .

接下来就经由 Netty 的 ChannelPipeline 机制, 将读取事件逐级发送到各个 handler 中, 于是就会触发前面我们提到的 ServerBootstrapAcceptor.channelRead 方法啦。

至此，分析了Netty线程池的初始化、ServerBootstrap的启动及channel与线程池的绑定过程，能够看出Netty中线程池的优雅设计，使用不同的线程池负责连接的建立、IO读写等，为API网关项目的高并发访问提供了技术基础。

总结

至此，对API网关技术的Netty实践分享就到这里，各位如果对中间的各个环节有什么疑问和建议，欢迎大家指正，我们一起讨论，共同学习提高。

参考

http://tutorials.jenkov.com/j...

http://netty.io/wiki/user-gui...

http://netty.io/

http://www.tuicool.com/articl...

作者：蜂巢团队 宜信技术学院