什么是异步
同步调用:调用方在调用过程中,持续等待返回结果。 异步调用:调用方在调用过程中,不直接等待返回结果,而是实行其他任务,结果返回形式常日为回调函数。
分开IO,单独谈论同步和异步,我们更加随意马虎去理解它的事理,同步和异步实在属于一种通信机制,表达的是,我们在通信过程中,是主动去讯问,还是等对方主动反馈。表示在同步(主动)和异步(被动)上。
进程和线程:进程是资源分配的最小单位,线程是CPU调度的最小单位

Java进程内最大略的异步调用办法,便是通过 new Thread().start() 的办法,启动新的线程进行任务的实行(CPU调度)。
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 的调用办法,属于同步非壅塞, 紧张缘故原由在于,在获取异步线程处理结果时,须要主线程主动去获取,异步线程并没有通过主动关照的办法,将数据构造进行更新或回调。
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
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、EventBuspublic 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("水开了!
2.1、不雅观察者模式
"); }catch (Exception e){ e.printStackTrace(); } }}
在 EventBus 中,通过 @Subscribe 定义了抽象不雅观察者的行为, 通过入口区分不同的事宜监听动作,如上述的示例代码中, listen(Event event) 只会不雅观察这个类的事宜。
/ 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() 方法,便利找到所有的监听方法:
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 正好供应了这个能力。
/ 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 的办理思路同等:
@Order(0)@Beanpublic ApplicationEventMulticaster applicationEventMulticaster() { SimpleApplicationEventMulticaster eventMulticaster = new SimpleApplicationEventMulticaster(); eventMulticaster.setTaskExecutor(Executors.newFixedThreadPool(10)); return eventMulticaster;}
在Spring 高下文初始化的时候,会将这一个bean,加载到高下文中,
存在的问题: 由于将全体高下文的 ApplicationEventMulticaster都更换了,那么在事宜处理的流程上,所有的事宜都会以异步的办法进行,那么风险的把控就很难做好。不建议,但能用(毕竟经受过磨练)
3.2、异步之@Async通过实现 AsyncConfigurer 接口,自定义线程池,对切面方法,实行反射代理
org.springframework.aop.interceptor.AsyncExecutionInterceptor#invoke
核心事理
进程间异步调用Dubbo 异步调用
在rpc框架中,我们普遍利用的都是同步调用模式,但是在 Dubbo 的底层实现中,反而因此 异步调用的办法实现的。先来大略看看调用链路:
核心代码在
com.alibaba.dubbo.rpc.protocol.dubbo.DubboInvoker#doInvoke
行列步队异步解耦
在先容 EventBus 的时候, 我查看了很多文章,都将EventBus 的设计模式 描述为发布-订阅 模式。首先这个描述是缺点的,然后我们来比拟一下他们的差异:
从表面上看:
不雅观察者模式里,只有两个角色 —— 不雅观察者 + 被不雅观察者而发布订阅模式里,却不仅仅只有发布者和订阅者两个角色,还有一个常常被我们忽略的 —— 经纪人Broker往更深层次讲:
不雅观察者和被不雅观察者,是松耦合的关系发布者和订阅者,则完备不存在耦合发布-订阅模式:
行列步队能够帮我们做到解耦的效果,通过中间件,如 RocketMQ,kafka和rabbitMQ等; 完成的吸收和推送,从而达到异步处理的效果。