Java Async IO Library: Quasar (use Fiber)

原创
2016/07/28 10:28
阅读数 1.7K

前言

Quasar 提供很多的功能(Go-like channels, Erlang-like actors),这篇博客主要介绍Quasar的核心Fiber和使用Fiber来处理异步IO(本博客给出的例子是发起Http 请求)。

本博客所指的线程均指Linux下的线程,Linux下不区分线程和进程,特别的地方会再做说明。

Fiber是什么?

Fiber(中文翻译成纤程) 是JVM上实现的轻量级用户态线程,和go 语言的goroutine 类似。Fiber 有如下几个特点:

  1. 用户态,用户态的线程切换是非常快的,而且单个时间片可以执行更多的代码(原因:操作系统何时进行线程调度(时间片用尽,IO中断,系统调用等))。众所周知,我们在进行系统调用(比如IO请求)的时候,当前的线程就会阻塞,产生线程上下文切换(从用户态切换到内核态)Context Switch,关于这个方面的测试请查看How long does it take to make a context switch?,从这里可以看出上下文切换的代价是非常高的,这也正是Fiber的优势。
  2. 轻量级,Fiber 占有的资源非常少,一个fiber大概在400 bytes of RAM,所以系统里可以同时有上百万个Fiber。

为什么使用Fiber(或者说使用Threads有什么问题)?

先看一下在JVM 中用户态线程和内核态线程的对应关系:

   Thread: 1:1 一个Java 线程对应一个内核线程.(可以被内核调度,消耗context switch)

   Fiber:  M:N mapping to kernel threads.

   Strand:  abstraction of thread or fiber(后面会有介绍).
  • 线程不是轻量的(heavy),也就是说单机能够产生的线程数量是非常有限的,不能满足现在Application并发的需求。
  • 在Java Web应用中,我们使用Java NIO 来接收TCP请求,线程和TCP连接是不匹配的。这种方式缺点是比较难以编码和维护(只是在Library 中使用,很少有用户代码中直接使用的),而Fiber 采取的方式是:"thread(fiber)-per-connection"

怎样不阻塞(non-block)?

传统做法:

  1. CallBacks(hell) 传统的回调,对应的问题就是末日金字塔...
public void messageFriend() {
       withModule(() -> {
                 withConnection(richard -> {
                        richard.dataHandler(data -> {
                                assertEquals("bob>oh its you!", data.toString());
                                moduleTestComplete();
                           });
                        richard.write("richard\n");
                        withConnection(bob -> {
                                       bob.dataHandler(data -> {
                                                                          assertEquals("richard>hai",data.toString());
                               bob.write("richard<oh its you!");
                               });
                        bob.write("bob\n");
                 vertx.setTimer(6, id -> richard.write("bob<hai"));
             });
         });
       });
}

2.Monads

In Java8 like:

  CompletableFuture.supplyAsync().thenAccept()...

这个要优于回调,它去除回调金字塔,但是也有如下的缺点,手动的上下文管理(需要在CompletableFuture 里面执行),并发逻辑不清晰(你会有很多的 .then().then()),还需要改变接口(方法需要返回 CompletableFuture)

你可以在这里看到更多的内容Monads vs Scoped Continuations

Fiber 的做法: Just Block, 因为Fiber 是轻量的,可以Suspend 和 Resume,Fiber 的执行过程是这样的,你创建并启动一个Fiber(Fiber创建和使用和线程一样):

new Fiber<Void>(new SuspendableRunnable() {
  public void run() throws SuspendExecution, InterruptedException {
    // your code
    bar(); // call bar;
  }
}).start();

然后由 Schedule(有默认提供) 调度,当Fiber需要block的时候,调用Fiber.park(),Schedule 可以执行其它的操作(效率就是在这个时候体现出来的),然后block完的时候 又通过 Fiber.unpark()继续执行。

输入图片说明

输入图片说明

Fiber在JVM上的实现方式(ByteCode instrumentation)

1.怎么 instrument:

Quasar fibers 依赖 bytecode instrumentation. 可以通过Java Agent在类加载的时候实现, 也可以通过在编译期间通过 Ant task来是实现. 实现的效果就是在原来的代码中插入一些额外的代码(或者说字节码)。

2.为什么Fiber是Suspendable 和 Resumeable 的,就是通过一个Stack来存储代码执行的相关信息:

class Stack{
    int[] method;  // PC(程序计数器), SP(栈指针)
    long[] dataLong; // stack premitives(基地址)
    Object[] dataObject; // stack refs (相关引用)
}

3.Instrumentation 之后在JVM中的执行过程:

before :

if bar() run in a fiber, and it call foo(),so foo() should be Suspendable.

bar(){
    baz();
    foo();          
}

foo(){
    Fiber.park(); // throw Exception
}

after:

bar(){
    int pc = isFiber ? s.pc :0;
    switch(pc){
        case 0:
            baz():
            if(isFiber){
                s.pc=1;
                // store locals -> s
            }
        case 1:
            if(isFiber)
                // load locals <-s
            foo(); // suspendable
    }            
}

foo(){
    int pc =isFiber ? s.pc : 0;
    switch(pc){
        if(isFiber){
            s.pc = 3;
            // store locals -> s
        }
        Fiber.park(); // throw Exception
    case 3:
        if(isFiber)
            // load locals <- s
    }
}

JVM上使用Fiber的一些问题

  • Interfaces/superclasses:iff have a suspendable implementation (dynamic proxies marked manually)
  • Reflection,invoke dynamic:presumed suspendable(except lambdas)
  • Which methods to instrument?(manual/graph analysis)
  • Instrument "infection":if we`re conservative and automatic,a lot of methods must be instrumented(e.g. consider Runnable being suspendable - and all it`s callers...)

这些问题的意思是说在进行Instrument的过程中,针对接口和动态代理方法,只有在运行时才能决定具体的实现类(决定具体的调用方法),所以需要手动管理列出这些类,这也是Fiber使用起来比较繁琐的一点,大家可以参考我最后给出的例子,例子中会给大家一个大概的说明(主要是为了新手少踩一些坑)。官方文档也给了详细的说明,需要耐心一点看完。

Fiber的使用场景

Fibers are not meant to replace threads in all circumstances. A fiber should be used when its body (the code it executes) blocks very often waiting on other fibers (e.g. waiting for messages sent by other fibers on a channel, or waiting for the value of a dataflow-variable). For long-running computations that rarely block, traditional threads are preferable. Fortunately, as we shall see, fibers and threads interoperate very well.

Fiber适用于阻塞频繁的代码(比如IO阻塞),而且如果需要消耗CPU的代码使用Thread,它们可以同时抽象为前面的Strand,这样的话优势就会高于Node(Node 通过单线程异步来高效实现IO请求处理,具体对比我会给出另外一篇博客)。

怎样开始?

建议感兴趣的朋友从官方文档开始Quasar

另外我做了一个使用Spring Boot 和 Fiber来做HttpClient的demo,放在GitHub上,以供大家参考 spring-quasar-demo

下一篇我会介绍在使用Fiber来做HttpServer。

展开阅读全文
打赏
3
1 收藏
分享
加载中
更多评论
打赏
0 评论
1 收藏
3
分享
OSCHINA
登录后可查看更多优质内容
返回顶部
顶部