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

1、背景

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

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

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

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

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

2、项目框架

图1 - API网关项目框架

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

3、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做为通讯模型，实现了基础框架的搭建。

4、Netty线程池

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

图6 – API网关线程池设计

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

• 先SelectKeys()处理NIO的事件

• 而后获取本线程的定时任务，放到本线程的任务队列里

• 最后执行其余线程提交给本线程的任务

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

5、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。

6、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用于处理网关业务加解密、限流、路由，及将请求转发给后端的抓取服务等业务操做。

7、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网关项目的高并发访问提供了技术基础。

8、总结

至此，对API网关技术的Netty实践分享就到这里，各位若是对中间的各个环节有什么疑问和建议，欢迎你们指正，咱们一块儿讨论，共同窗习提升。

参考

http://tutorials.jenkov.com/java-nio/nio-vs-io.html

http://netty.io/wiki/user-guide-for-4.x.html

http://netty.io/

http://www.tuicool.com/articles/mUFnqeM

https://segmentfault.com/a/1190000007403873

https://segmentfault.com/a/1190000007283053

做者：蜂巢团队 

来源：宜信技术学院