《高性能利器》--异步调用实现原理详解！

概述

高可用系列文章 算是告一段落了，主要讲了 限流，熔断，削峰和 Sentinel实战 ：

《高可用系列》-- 你来说说什么是限流？
《高可用系列》-- 限流神器Sentinel，不了解一下吗？
《高可用系列》-- 阿里P7大佬带你解密Sentinel
《高可用系列》-- 熔断降级我学会了！
《高可用系列》--来聊聊削峰填谷！

接下来进入到大家比较喜欢的 高性能系列，主题内容包括，消息队列，缓存和分布式部署架构等，在上一篇文章- # 秒杀系统架构图该怎么画？手把手教你！ ,讲解了博主凄惨的经历，因此在学习相关技术的时候，我们要将其运用到我们实际的项目中，在高性能篇结束后，将进入架构图2.0版本～

什么是异步

代码语言：javascript复制

同步调用：调用方在调用过程中，持续等待返回结果。
 
异步调用：调用方在调用过程中，不直接等待返回结果，而是执行其他任务，结果返回形式通常为回调函数。

脱离IO，单独讨论同步和异步,我们更加容易去理解它的原理，同步和异步其实属于一种通信机制，表达的是，我们在通信过程中，是主动去询问，还是等对方主动反馈。体现在同步（主动）和异步（被动）上。

进程内异步调用

1、Thread

进程和线程：进程是资源分配的最小单位，线程是CPU调度的最小单位

Java进程内最简单的异步调用方式，就是通过 new Thread().start() 的方式，启动新的线程进行任务的执行（CPU调度）。

代码语言：javascript复制

public static void main(String[] args) {
    System.out.println("煲水");
    //创建新的线程
    Thread thread1= new Thread(()->{
        try {
            Thread.sleep(5000);
            System.out.println("水开了," Thread.currentThread().getName());
        }catch (Exception e){
            e.printStackTrace();
        }
    });
    thread1.start();
    System.out.println("运动");
}

1.1、start() 和 run()

在上述实例代码中，我们虽然采用了实现 Runnable 接口的方式，进行新线程的实现，但是在方法启动时，并没有使用 run() 方法，而是使用了 start() 方法。

run（）：使用当前线程执行 run（）方法调用，可以理解时同步调用

start（） 方法在调用时，在代码逻辑中，会调用到一个本地方法 start0，

下载 JDK源码后，可以看到 Thread 类有个 registerNatives 本地方法，该方法主要的作用就是注册一些本地方法供 Thread 类使用，如 start0()，stop0() 等等，可以说，所有操作本地线程的本地方法都是由它注册的。

可以看出 Java 线程 调用 start->start0 的方法，实际上会调用到 JVM_StartThread 方法，通过调用 new JavaThread(&thread_entry,sz) 完成线程的创建。

在 jvm.cpp 中，有如下代码段：

在创建完线程后，通过 thread_entry 完成 run() 方法的调用

1.2、Future

Future 的调用方式，属于同步非阻塞, 主要原因在于，在获取异步线程处理结果时，需要主线程主动去获取，异步线程并没有通过主动通知的方式，将数据结构进行更新或回调。

代码语言：javascript复制

public static void main(String[] args) throws Exception {
    System.out.println("煲水");
    FutureTask<String> futureTask = new FutureTask(()->{
        try {
            Thread.sleep(5000);
            System.out.println("水开了," Thread.currentThread().getName());
        }catch (Exception e){
            e.printStackTrace();
        }
        return "water";
    });
    //创建新的线程
    Thread thread1= new Thread(futureTask);
    thread1.start();
    System.out.println("运动");
    Thread.sleep(3000);
    //阻塞等待数据
    String result= futureTask.get(5, TimeUnit.SECONDS);
    System.out.println("喝水,"   result);
}

Future 的实现原理

类的继承关系图如下，可以看到 FutureTask，实现了 Runnable 接口,那么在重写的run（） 方法中，可以看到，在调用 call（） 方法获取到结果后，通过CAS的方式，更新到成员变量中。

任务调用结果更新：

1.3、ThreadPoolExecutor

代码语言：javascript复制

public static void main(String[] args) throws Exception {
    ExecutorService executors = Executors.newFixedThreadPool(10);
    System.out.println("煲水");
    Future<String> future = executors.submit(() -> {
        try {
            Thread.sleep(5000);
            System.out.println("水开了,"   Thread.currentThread().getName());
        } catch (Exception e) {
            e.printStackTrace();
        }
        return "water";
    });
    System.out.println("运动");

    String result = future.get(5, TimeUnit.SECONDS);
    System.out.println("喝水,"   result);
}

上面讲解了 FutureTask 的实现原理后，这里在对比 submit() 和 execute()，就比较容易理解了，在submit()方法中，将 Callable<T> 实现类，封装成了 FutureTask，然后再进行实际的调用:

1.4、总结

核心区别在于 start() 和 run() ， start（）是启动一条新的线程的同时，完成run()方法，这时候是一个异步操作；如果直接执行run()方法，则会在当前线程直接执行,是一个同步阻塞操作。

而 Future 的调用方式，则是一个 同步非阻塞处理，在提交了任务后,不阻塞主线程的继续执行，在到了一定时间后，主线程可以通过get（） 方法，获取异步任务处理结果。

ThreadPoolExecutor 则是维护了一个可复用的线程池，解决了资源复用，性能耗时的问题, Java线程默认大小为1MB,线程的创建和销毁都会占用内存和GC耗时；而线程的无限制创建，则会带来CPU负载过高,每个线程分配的时间片很少，导致处理效率低。

2、EventBus

代码语言：javascript复制

public class JiulingTest {
    public static void main(String[] args) throws Exception {
        System.out.println("开始");
        //使用异步事件总线
        EventBus eventBus  = new AsyncEventBus(Executors.newFixedThreadPool(10));
        // 向上述EventBus对象中注册一个监听对象
        eventBus.register(new EventListener());
        // 使用EventBus发布一个事件，该事件会给通知到所有注册的监听者
        eventBus.post(new Event("煲水"));
        System.out.println("运动");
    }
}
// 事件，监听者监听的事件的包装对象
class Event {
    //事件动作
    public String action;

    Event(String action) {
        this.action = action;
    }
}

// 监听者
 class EventListener {
    // 监听的方法，必须使用注解声明，且只能有一个参数，实际触发一个事件的时候会根据参数类型触发方法
    @Subscribe
    public void listen(Event event) {
        try {
            System.out.println("Event listener receive message:  "   event.action   " threadName:"   Thread.currentThread().getName());
            Thread.sleep(5000);
            System.out.println("水开了！");
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

2.1、观察者模式

在 EventBus 中，通过 @Subscribe 定义了抽象观察者的行为, 通过入参区分不同的事件监听动作，如上述的示例代码中， listen(Event event) 只会观察这个类的事件。

代码语言：javascript复制

/**
 * Returns all subscribers for the given listener grouped by the type of event they subscribe to.
 */
private Multimap<Class<?>, Subscriber> findAllSubscribers(Object listener) {
  Multimap<Class<?>, Subscriber> methodsInListener = HashMultimap.create();
  Class<?> clazz = listener.getClass();
  //遍历 @Subscribe 的方法
  for (Method method : getAnnotatedMethods(clazz)) {
    Class<?>[] parameterTypes = method.getParameterTypes();
    Class<?> eventType = parameterTypes[0];
    //然后根据 参数类型，也就是事件类型，进行归类
    methodsInListener.put(eventType, Subscriber.create(bus, listener, method));
  }
  return methodsInListener;
}

然后在进行事件发布的时候，通过调用 EventBus.post（） 方法，遍历找到所有的监听方法：

代码语言：javascript复制

public void post(Object event) {
//从上述归类的Map 中，找到所有的观察者方法
  Iterator<Subscriber> eventSubscribers = subscribers.getSubscribers(event);
  if (eventSubscribers.hasNext()) {
  //事件分发，具体调用
    dispatcher.dispatch(event, eventSubscribers);
  } else if (!(event instanceof DeadEvent)) {
    // the event had no subscribers and was not itself a DeadEvent
    post(new DeadEvent(this, event));
  }
}

2.2、AsyncEventBus

在示例代码中，我们使用的是 new AsyncEventBus(Executors.newFixedThreadPool(10)) 构建的异步事件总线。

由下往上倒推，我们先看 Listern ，是如何执行事件处理方法的，这里比较好理解，通过线程池完成任务的调用，具体实现是通过反射的方式调用 @Subscribe 注解的方法。

那么这里的 executor 是怎么来的呢？

this.executor = bus.executor(); //从事件总线传递过来

回到 EventBus 中，我们可以看到构造函数并没有提供初始化线程池的入口，那么默认线程池的创建，可以跟踪到

这个线程池的 execute 方法，并没有创建新的线程执行 Runnable 方法，而是使用当前线程执行（具体逻辑参考1.1）。因此 EventBus 是不支持异步事件处理的！

在 dispatchEvent 方法中，比较直接可以看到整体设计中，是支持异步事件的，我们需要做的就是将 Executor 修改成一个合理的线程池, 而 AsyncEventBus 恰恰提供了这个能力。

代码语言：javascript复制

/**
 * Creates a new AsyncEventBus that will use {@code executor} to dispatch events.
 *
 * @param executor Executor to use to dispatch events. It is the caller's responsibility to shut
 *     down the executor after the last event has been posted to this event bus.
 */
public AsyncEventBus(Executor executor) {
  super("default", executor, Dispatcher.legacyAsync(), LoggingHandler.INSTANCE);
}

3、Spring Event

Spring Event 与 Event Bus 默认都是同步执行，支持通过设置 Executors 的方式修改成异步事件。

核心组件：

事件类：定义事件，继承ApplicationEvent的类成为一个事件类。
发布者：发布事件，通过ApplicationEventPublisher发布事件。
监听者：监听并处理事件，实现ApplicationListener接口或者使用@EventListener注解。

由于代码过多，可以直接github 下载进行阅读，这里只贴部分关键代码:

在发布事件方法：AbstractApplicationContext#publishEvent 会走到下图中的 SimpleApplicationEventMulticaster#multicastEvent 执行具体任务的调度。这里的设计与上面的 EventBus 如出一辙,在执行时，通过区分线程池进行实际的调度，从而决定 同步|异步!

3.1、异步之ApplicationEventMulticaster

修改 ApplicationEventMulticaster 设置初始线程池，和EventBus 的解决思路一致:

代码语言：javascript复制

@Order(0)
@Bean
public ApplicationEventMulticaster applicationEventMulticaster() {
    SimpleApplicationEventMulticaster eventMulticaster = new SimpleApplicationEventMulticaster();
    eventMulticaster.setTaskExecutor(Executors.newFixedThreadPool(10));
    return eventMulticaster;
}

在Spring 上下文初始化的时候，会将这一个bean，加载到上下文中，