任丘建设网站制作,安阳网站,推广策略分析,有什么网站做交流会总结自尼恩的全链路异步#xff1a;网关纯异步化网关层的特点#xff1a;不需要访问业务数据库只做协议转换和流量转发特点是 IO 密集型#xff0c;特别适合纯异步的架构#xff0c;可以极大的节省资源。如何进行网关异步化#xff1f;使用高性能的通信框架Netty#xff…总结自尼恩的全链路异步网关纯异步化网关层的特点不需要访问业务数据库只做协议转换和流量转发特点是 IO 密集型特别适合纯异步的架构可以极大的节省资源。如何进行网关异步化使用高性能的通信框架Netty这是一个基于NIO 非阻塞IO Reactor 纯异步线程模型的纯异步化框架springcloud getway 它是基于spring 5.0 、spring boot 2.0 和spring reacter为微服务提供一个简单有效的网关API路由接口。和SpringCloud 框架完美集成目标是为了代替zuulSpringCloud GetWay 是基于webFlux框架实现的而WebFlux框架底层则使用了高性能的Reactor模式通信框架Netty。所以最终还是基于IO的王者组件Netty。Web 服务异步化2W并发场景提升 20倍以上Springboot的Web 服务默认为 Tomcat Servlet 不支持纯异步化编程Tomcat Servlet模式的问题总体上没有使用Reactor 反应器模式 每一个请求是阻塞处理的属于同步 Web 服务类型。Servlet 有异步的版本可惜没有用起来。如何实现 Web 服务异步化方式一基于 Netty 实现web服务方式二使用 WebFlux 还是 Netty 实现web服务Spring WebFlux是一个响应式堆栈 Web 框架 它是完全非阻塞的支持响应式流(Reactive Stream)背压并在NettyUndertow和Servlet 3.1 容器等服务器上运行RPC 调用异步化异步RPC 调用等待upstream 上游 response 返回时线程不处于block 状态作为微服务架构中数据流量最大的一部分RPC 调用异步化的收益巨大RPC 调用主要的框架有特点是feign 是同步IO 、阻塞模式的同步 RPC框架dubbo 是基于Netty的非阻塞IO Reactor 反应堆线程模型的 异步RPC框架从数据来看 dubbo rpc 是feign rpc 性能10倍Cache异步化提升2倍Cache Aside 缓存模式是大家通用的Cache使用方式Cache纯异步的架构必须使用异步存储层客户端主要有RedissonLettuceRedisson、Lettuce如何选型请参考40岁老架构师尼恩的文章使用建议建议lettuce Redisson在spring boot2之后redis连接默认就采用了lettuce。就像 spring 的本地缓存默认使用Caffeine一样在Spring Boot 2.0中将取代GuavaCaffeine是使用Java8对Guava缓存的重写版本。性能达到 400W qps非常牛逼。这就一定程度说明了lettuce 比 Jedis在性能的更加优秀。DB的异步化 数据操作是每个请求调用链的 终点纯异步的架构必须使用异步存储层客户端比如说可以使用纯异步化的框架 Spring Data R2DBCDB是一个低吞吐的物种对于DB而已请求太多反而忙不过来造成整体的性能下降。参考 Hystrix 舱壁模式 通过 DB 的操作进行 线程池隔离使用 手写 Hystrix Command 的方式进行 DB 操作的 高压防护。控制线程数和请求数保护不至于拖垮DB纯异步与伪异步异步调用目的在于防止当前业务线程被阻塞。伪异步将任务包装为Runnable 放入另一个线程执行并等待当前Biz 线程不阻塞纯异步为响应式编程模型通过IO 实践驱动任务完成。使用到异步的场景场景1 超高并发 批量 写 mysql 、批量写 elasticSearch场景2 超高并发 批量 IO场景3 超高并发 发送短信、发邮件场景4 超高并发 发送消息场景5 超高吞吐 生产者、 消费者 场景场景6 超高吞吐 发布、 订阅 场景场景7 分布式的 通知场景场景8异步回调场景场景9其他的 异步场景 不知道能列举出多少总之非常多首先、什么是异步同步调用方在调用过程中持续阻塞一直到返回结果。同步获取结果的方式是 主动等待。异步调用方在调用过程中不会阻塞 不直接等待返回结果 而是执行其他任务。异步获取结果的方式是 被动通知或者 被动回调。异步的20种实现方式新建线程Thread 实现异步线程池化 实现异步Future 阻塞式异步guava 回调式异步Netty 回调式异步Servlet 3.0 异步CompletableFuture 回调式异步JDK 9 Flow 响应式编程RxJava 响应式 异步Reactor 响应式 异步Spring注解Async 异步EventBus 框架 发布订阅模式异步Spring ApplicationEvent 事件 发布订阅模式RocketMq 消息队列 分布式 发布订阅模式Pub/Sub 异步Redis 消息队列 分布式 发布订阅模式Pub/Sub 异步Distruptor 框架异步ForkJoin 框架异步RocketMQ源码中ServiceThread 能急能缓的高性能异步Kotlin 协程 异步Project Loom 协程 异步123没什么好说的基操4是继承3没什么好会说的。Netty异步我们知道JDK的Future的任务结果获取需要主动查询而Netty的Future通过添加监听器Listener可以做到异步非阻塞处理任务结果可以称为被动回调。本质上没什么特别高深的。就是for循环便利listenler实现逻辑上的解耦并使用一个或者少量线程资源即处理。具体参考此文https://zhuanlan.zhihu.com/p/385350359Callback Hell回调地狱问题无论是 Google Guava 包中的 ListenableFuture还是 Netty的 GenericFutureListener都是需要设置专门的Callback 回调钩子Guava 包中的 ListenableFuture设置Callback 回调钩子的实例如下ListenableFutureBoolean wFuture gPool.submit(wJob);Futures.addCallback(wFuture, new FutureCallbackBoolean() {public void onSuccess(Boolean r) {if (!r) {Print.tcfo(杯子洗不了没有茶喝了);} else {countDownLatch.countDown();}}public void onFailure(Throwable t) {Print.tcfo(杯子洗不了没有茶喝了);}});调用方通过 Futures.addCallback() 添加处理结果的回调函数。这样避免获取并处理异步任务执行结果阻塞调起线程的问题。Callback 是将任务执行结果作为接口的入参在任务完成时回调 Callback 接口执行后续任务从而解决纯 Future 方案无法方便获得任务执行结果的问题。但 Callback 产生了新的问题那就是代码可读性的问题。因为使用 Callback 之后代码的字面形式和其所表达的业务含义不匹配即业务的先后关系到了代码层面变成了包含和被包含的关系。因此如果大量使用 Callback 机制将使大量的存在先后次序的业务逻辑在代码形式上转换成层层嵌套从而导致业务先后次序在代码维度被打乱最终造成代码不可理解、可读性差、难以理解、难以维护。这便是所谓的 Callback Hell回调地狱问题。Callback Hell 问题可以从两个方向进行一定的解决一是链式调用二是事件驱动机制。前被 CompletableFuture、反应式编程等技术采用前者被如 EventBus、Vert.x 所使用。CompletableFuturehttps://baijiahao.baidu.com/s?id1739828295604310156wfrspiderforpc反应式编程https://blog.csdn.net/qq_42799615/article/details/111235576Vert.x https://blog.csdn.net/wjw465150/article/details/125016949Servlet 3.0 异步Callback 真正体现价值是它与 NIO 技术结合之后。CPU 密集型场景采用 Callback 回调没有太多意义IO 密集型场景如果是使用 BIO模式Callback 同样没有意义因为一个连接一个线程IO线程是因为 IO 而阻塞。IO 密集型场景如果是使用 NIO 模式使用Callback 才有意义。 NIO是少量IO线程负责大量IO通道IO线程需要避免线程阻塞所以也必须使用 Callback 才能使应用得以被开发出来。所以高性能的 NIO 框架如 Netty 都是基于 Callback 异步回调的。微服务系统中多级服务调用很常见一个服务先调 A再用结果 A 调 B然后用结果 B 调用 C等等。如果使用Netty 作为底层服务器IO 线程能大大降低能处理的连接数/请求数也能大大增加那么为啥Netty 却没有在WEB服务器中占据统治地位呢这其中的难度来自两方面一是 NIO 和 Netty 本身的技术难度二是 Callback hellCallback 风格所导致的代码理解和维护的困难。所以Servlet 3.0 提供了一个异步解决方案。什么是servlet异步请求https://blog.csdn.net/qq_35789269/article/details/127974069?ops_request_misc%257B%2522request%255Fid%2522%253A%2522167716485916800211525010%2522%252C%2522scm%2522%253A%252220140713.130102334.pc%255Fblog.%2522%257Drequest_id167716485916800211525010biz_id0utm_mediumdistribute.pc_search_result.none-task-blog-2~blog~first_rank_ecpm_v1~rank_v31_ecpm-5-127974069-null-null.blog_rank_defaultutm_termseverletspm1018.2226.3001.4450Servlet 3.0 之前一个普通 Servlet 的主要工作流程大致如下1Servlet 接收到请求之后可能需要对请求携带的数据进行一些预处理2调用业务接口的某些方法以完成业务处理3根据处理的结果提交响应Servlet 线程结束。其中第二步处理业务逻辑时候很可以碰到比较耗时的任务此时servlet主线程会阻塞等待完成业务处理对于并发比较大的请求可能会产生性能瓶颈则servlet3.0之后再此处做了调整引入了异步的概念。1Servlet 接收到请求之后可能需要对请求携带的数据进行一些预处理2调用业务接口的某些方法过程中request.startAsync()请求获取一个AsyncContext3紧接着servlet线程退出(回收到线程池)但是响应response对象仍旧保持打开状态新增线程会使用AsyncContext处理并响应结果。4AsyncContext处理完成触发某些监听通知结果 WebServlet(urlPatterns /demo, asyncSupported true)public class AsyncDemoServlet extends HttpServlet {Overridepublic void doGet(HttpServletRequest req, HttpServletResponse resp) throws IOException, ServletException {
// Do SomethingAsyncContext ctx req.startAsync();startAsyncTask(ctx);}}private void startAsyncTask(AsyncContext ctx) {requestRpcService(result - {try {PrintWriter out ctx.getResponse().getWriter();out.println(result);out.flush();ctx.complete();} catch (Exception e) {e.printStackTrace();}});}
Servlet 3.0 的出现解决了在过去基于 Servlet 的 Web 应用中接受请求和返回响应必须在同一个线程的问题实现了如下目标可以避免了 Web 容器的线程被阻塞挂起使请求接收之后的任务处理可由专门线程完成不同任务可以实现线程池隔离结合 NIO 技术实现更高效的 Web 服务除了直接使用 Servlet 3.0也可以选择 Spring MVC 的 Deferred Result。 GetMapping(/async-deferredresult)public DeferredResultResponseEntity? handleReqDefResult(Model model) {LOG.info(Received async-deferredresult request);DeferredResultResponseEntity? output new DeferredResult();ForkJoinPool.commonPool().submit(() - {LOG.info(Processing in separate thread);try {Thread.sleep(6000);} catch (InterruptedException e) {}output.setResult(ResponseEntity.ok(ok));});LOG.info(servlet thread freed);return output;}
Servlet 3.0 的技术局限Servlet 3.0 并不是用来解决前面提到的 Callback Hell 问题的它只是降低了异步 Web 编程的技术门槛。对于 Callback Hell 问题使用 Servlet 3.0 或类似技术时同样会遇到。解决 Callback Hell 还需另寻他法。回调式 异步CompletableFutureJDK 1.8之前并没有实现回调式的异步CompletableFuture是JDK 1.8引入的实现类实现了JDK内置的异步回调模式异步。CompletableFuture的创新是通过 链式调用解决 Callback Hell回调地狱问题 让代码变得的可理解行更强可读性 更强。CompletableFuture 该类实现了Future和CompletionStage两个接口。该类的实例作为一个异步任务可以在自己异步执行完成之后触发一些其他的异步任务从而达到异步回调的效果。package com.crazymakercircle.completableFutureDemo;import com.crazymakercircle.util.Print;import java.util.concurrent.CompletableFuture;import static com.crazymakercircle.util.ThreadUtil.sleepSeconds;public class DrinkTea {private static final int SLEEP_GAP 3;//等待3秒public static void main(String[] args) {// 任务 1CompletableFutureBoolean washJob CompletableFuture.supplyAsync(() -{Print.tcfo(洗茶杯);//线程睡眠一段时间代表清洗中sleepSeconds(SLEEP_GAP);Print.tcfo(洗完了);return true;});// 任务 2CompletableFutureBoolean hotJob CompletableFuture.supplyAsync(() -{Print.tcfo(洗好水壶);Print.tcfo(烧开水);//线程睡眠一段时间代表烧水中sleepSeconds(SLEEP_GAP);Print.tcfo(水开了);return true;});// 任务 3任务 1 和任务 2 完成后执行泡茶CompletableFutureString drinkJob washJob.thenCombine(hotJob, (hotOk, washOK) -{if (hotOk washOK) {Print.tcfo(泡茶喝茶喝完);return 茶喝完了;}return 没有喝到茶;});// 等待任务 3 执行结果Print.tco(drinkJob.join());}
}JDK 9 Flow 响应式编程但是 JDK 8 的 CompletableFuture 属于链式调用它在形式上带有一些响应式编程的函数式代码风格。因为 Callback Hell 对代码可读性有很大杀伤力从开发人员的角度来讲反应式编程技术和链式调用一样使得代码可读性要比 Callback 提升了许多。响应式流从2013年开始作为提供非阻塞背压的异步流处理标准的倡议。Reactive Stream是一套基于发布/订阅模式的数据处理规范。更确切地说Reactive流目的是“找到最小的一组接口方法和协议用来描述必要的操作和实体以实现这样的目标以非阻塞背压方式实现数据的异步流”。**响应式流Reactive Streams**是一个响应式编程的规范用来为具有非阻塞背压Back pressure的异步流处理提供标准用最小的一组接口、方法和协议用来描述必要的操作和实体。这里涉及到一个关键概念叫 Backpressure国内大部分翻译为背压我们先来了解这是什么。响应式编程其实就是对数据流的编程而对流的处理对数据流的变化进行响应是通过异步监听的方式来处理的。既然是异步监听就涉及到监听事件的发布者和订阅者数据流其实就是由发布者生产再由一个或多个订阅者进行消费的元素item序列。那么如果发布者生产元素的速度和订阅者消费元素的速度不一样是否会出现问题呢其实就两种情况发布者生产的速度比订阅者消费的速度慢那生产的元素可以及时被处理订阅者处理完只要等待发布者发送下一元素即可这不会产生什么问题。发布者生产的速度比订阅者消费的速度快那生产的元素无法被订阅者及时处理就会产生堆积如果堆积的元素多了订阅者就会承受巨大的资源压力(pressure)而有可能崩溃。要应对第二种情况就需要进行流控制flow control。流控制有多种方案其中一种机制就是 Back pressure即背压机制其实就是下游能够向上游反馈流量需求的机制。如果生产者发出的数据比消费者能够处理的数据量大而且快时消费者可能会被迫一直再获取或处理数据消耗越来越多的资源从而埋下潜在的风险。为了防止这一点消费者可以通知生产者降低生产数据的速度。生产者可以通过多种方式来实现这一要求这时候我们就会用到背压机制。采用背压机制后消费者告诉生产者降低生产数据速度并保存元素知道消费者能够处理更多的元素。使用背压可以有效地避免过快的生产者压制消费者。如果生产者要一直生产和保存元素使用背压也可能会要求其拥有无限制的缓冲区。生产者也可以实现有界缓冲区来保存有限数量的元素如果缓冲区已满可以选择放弃。背压的实现方式背压的实现方式有两种一种是阻塞式背压另一种是非阻塞式背压。阻塞式背压阻塞式背压是比较容易实现的例如当生产者和消费者在同一个线程中运行时其中任何一方都将阻塞其他线程的执行。这就意味着当消费者被执行时生产者就不能发出任何新的数据元素。因而也需要一中自然地方式来平衡生产数据和消费数据的过程。在有些情况下阻塞式背压会出现不良的问题比如当生产者有多个消费者时不是所有消费者都能以同样的速度消费消息。当消费者和生产者在不同环境中运行时这就达不到降压的目的了。2、非阻塞式背压背压机制应该以非阻塞式的方式工作实现非阻塞式背压的方法是放弃推策略采用拉策略。生产者发送消息给消费者等操作都可以保存在拉策略当中消费者会要求生产者生成多少消息量而且最多只能发送这些量然后等到更多消息的请求。JDK 8 的 CompletableFuture 不算是反应式编程不遵守Reactive Stream (响应式流/反应流) 规范。JDK 9 Flow 是JDK对Reactive Stream (响应式流/反应流) 的实现Java 9中新增了反应式/响应式编程的Api-Flow。在Flow中存在Publisher发布者、Subscriber订阅者、Subscription订阅和Processor处理器。JDK 9 Flow 旨在解决处理元素流的问题——如何将元素流从发布者传递到订阅者而不需要发布者阻塞或订阅者需要有无限制的缓冲区或丢弃。当然实施响应式编程需要完整的解决方案单靠 Flow 是不够的还是需要Netflix RxJava、 Project Reactor 这样的完整解决方案。但是 JDK 层面的技术能提供统一的技术抽象和实现在统一技术方面还是有积极意义的。所以这里不对 JDK 9 Flow 做介绍。使用 RxJava 基于事件流的链式调用、代码 逻辑清晰简洁。package com.crazymakercircle.completableFutureDemo;import com.crazymakercircle.util.Print;
import io.reactivex.Observable;
import io.reactivex.schedulers.Schedulers;
import org.junit.Test;import java.util.concurrent.CompletableFuture;import static com.crazymakercircle.util.ThreadUtil.sleepMilliSeconds;
import static com.crazymakercircle.util.ThreadUtil.sleepSeconds;public class IntegrityDemo {/*** 模拟模拟RPC调用1*/public String rpc1() {//睡眠400ms,模拟执行耗时sleepMilliSeconds(600);Print.tcfo(模拟RPC调用服务器 server 1);return sth. from server 1;}/*** 模拟模拟RPC调用2*/public String rpc2() {//睡眠400ms,模拟执行耗时sleepMilliSeconds(600);Print.tcfo(模拟RPC调用服务器 server 2);return sth. from server 2;}Testpublic void rpcDemo() throws Exception {CompletableFutureString future1 CompletableFuture.supplyAsync(() -{return rpc1();});CompletableFutureString future2 CompletableFuture.supplyAsync(() - rpc2());CompletableFutureString future3 future1.thenCombine(future2,(out1, out2) -{return out1 out2;});String result future3.get();Print.tco(客户端合并最终的结果 result);}Testpublic void rxJavaDemo() throws Exception {ObservableString observable1 Observable.fromCallable(() -{return rpc1();}).subscribeOn(Schedulers.newThread());ObservableString observable2 Observable.fromCallable(() - rpc2()).subscribeOn(Schedulers.newThread());Observable.merge(observable1, observable2).observeOn(Schedulers.newThread()).toList().subscribe((result) - Print.tco(客户端合并最终的结果 result));sleepSeconds(Integer.MAX_VALUE);}}Reactor 响应式 异步Reactor 是一个运行在 Java8 之上满足 Reactice 规范的响应式框架它提供了一组响应式风格的 API。Reactor 有两个核心类 FluxT 和 MonoT这两个类都实现 Publisher 接口。Flux 类似 RxJava 的 Observable它可以触发零到多个事件并根据实际情况结束处理或触发错误。Mono 最多只触发一个事件所以可以把 Mono 用于在异步任务完成时发出通知。Spring的Async异步在Spring中使用Async标注某方法可以使该方法变成异步方法这些方法在被调用的时候将会在独立的线程中进行执行调用者不需等待该方法执行完成。但在Spring中使用Async注解需要使用EnableAsync来开启异步调用。一个 AsyncService参考代码如下public interface AsyncService {MessageResult sendSms(String callPrefix, String mobile, String actionType, String content);MessageResult sendEmail(String email, String subject, String content);
}Slf4j
Service
public class AsyncServiceImpl implements AsyncService {Autowiredprivate IMessageHandler mesageHandler;OverrideAsync(taskExecutor)public MessageResult sendSms(String callPrefix, String mobile, String actionType, String content) {try {Thread.sleep(1000);mesageHandler.sendSms(callPrefix, mobile, actionType, content);} catch (Exception e) {log.error(发送短信异常 - , e)}}OverrideAsync(taskExecutor)public sendEmail(String email, String subject, String content) {try {Thread.sleep(1000);mesageHandler.sendsendEmail(email, subject, content);} catch (Exception e) {log.error(发送email异常 - , e)}}
}
Async注解默认使用系统自定义线程池。在实际项目中推荐等方式是是使用自定义线程池的模式。可在项目中设置多个线程池在异步调用的时候指明需要调用的线程池名称比如Async(“taskName”)自定义异步线程池的代码如下/*** 线程池参数配置多个线程池实现线程池隔离
EnableAsync
Configuration
public class TaskPoolConfig {/*** 自定义线程池***/Bean(taskExecutor)public Executor taskExecutor() {//返回可用处理器的Java虚拟机的数量 12int i Runtime.getRuntime().availableProcessors();System.out.println(系统最大线程数 i);ThreadPoolTaskExecutor executor new ThreadPoolTaskExecutor();//核心线程池大小executor.setCorePoolSize(16);//最大线程数executor.setMaxPoolSize(20);//配置队列容量默认值为Integer.MAX_VALUEexecutor.setQueueCapacity(99999);//活跃时间executor.setKeepAliveSeconds(60);//线程名字前缀executor.setThreadNamePrefix(asyncServiceExecutor -);//设置此执行程序应该在关闭时阻止的最大秒数以便在容器的其余部分继续关闭之前等待剩余的任务完成他们的执行executor.setAwaitTerminationSeconds(60);//等待所有的任务结束后再关闭线程池executor.setWaitForTasksToCompleteOnShutdown(true);return executor;}
}
EventBus 发布订阅模式异步实际开发中常常 通过事件总线EventBus/AsyncEventBus进行JAVA模块解耦 比如在顶级开源组件 JD hotkey的源码中 就多次用到 EventBus/AsyncEventBus进行JAVA模块解耦掌握了 EventBus 在平时的开发中多了一个神器EventBus 是 Guava 的事件处理机制是观察者模式生产/消费模型的一种实现。EventBus是google的Guava库中的一个处理组件间通信的事件总线它基于发布/订阅模式实现了多组件之间通信的解耦合事件产生方和事件消费方实现解耦分离提升了通信的简洁性。观察者模式在我们日常开发中使用非常广泛例如在订单系统中订单状态或者物流信息的变更会向用户发送APP推送、短信、通知卖家、买家等等审批系统中审批单的流程流转会通知发起审批用户、审批的领导等等。Observer模式也是 JDK 中自带就支持的其在 1.0 版本就已经存在 Observer不过随着 Java 版本的飞速升级其使用方式一直没有变化许多程序库提供了更加简单的实现例如 Guava EventBus、RxJava、EventBus 等为什么要用 EventBus 其优点 EventBus 优点相比 Observer 编程简单方便通过自定义参数可实现同步、异步操作以及异常处理单进程使用无网络影响缺点只能单进程使用项目异常重启或者退出不保证消息持久化如果需要分布式使用还是需要使用 MQ在工作中经常会遇见使用异步的方式来发送事件或者触发另外一个动作经常用到的框架是MQ分布式方式通知; 如果是同一个jvm里面通知的话就可以使用EventBus 事件总线。由于EventBus使用起来简单、便捷因此工作中会经常用到。EventBus 是线程安全的分发事件到监听器并提供相应的方式让监听器注册它们自己。EventBus允许组件之间进行 “发布-订阅” 式的通信而不需要这些组件彼此知道对方。EventBus是专门设计用来替代传统的Java进程内的使用显示注册方式的事件发布模式。EventBus不是一个通用的发布-订阅系统也不是用于进程间通信。EventBus有三个关键要素具体可参考此文https://blog.csdn.net/crazymakercircle/article/details/128627663Spring ApplicationEvent事件实现异步Spring内置了简便的事件机制原理和EventBus 差不多通过Spring ApplicationEvent事件 可以非常方便的实现事件驱动核心类包括ApplicationEvent具体事件内容事件抽象基类可继承该类自定义具体事件ApplicationEventPublisher事件发布器可以发布ApplicationEvent也可以发布普通的Object对象ApplicationListener事件监听器可以使用注解EventListenerTransactionalEventListener事务事件监听可监听事务提交前、提交后、事务回滚、事务完成成功或失败使用示例不定义事件直接发布Object对象同步定义发送事件对象public class UserEntity {private long id;private String name;private String msg;
}2、定义事件监听器可以添加条件condition限制监听具体的事件Slf4j
Component
public class RegisterListener {EventListener(condition #entity.id ! null and #entity.asyncfalse )public void handlerEvent(UserEntity entity) {try {// 休眠5秒TimeUnit.SECONDS.sleep(5);} catch (InterruptedException e) {e.printStackTrace();}log.info(handlerEvent: {}, entity);}}3、定义发送接口以及实现类public interface IRegisterService {public void register(String name);}
12345
Service
public class RegisterServiceImpl implements IRegisterService {Resourceprivate ApplicationEventPublisher applicationEventPublisher;Overridepublic void publish(String name) {UserEntity entity new UserEntity();entity.setName(name);entity.setId(1L);entity.setMsg(新用户注册同步调用);applicationEventPublisher.publishEvent(entity);}
}4、测试Controller类进行测试Slf4j
Controller
public class TestController {Resourceprivate IRegisterService registerService;RequestMapping(test)ResponseBodypublic void test1(String name) {registerService.publish(name);log.info(执行同步调用结束);}
}更多介绍可参考此文Spring注解 EventListener 的介绍与使用示例以及异常处理https://blog.csdn.net/justyuze/article/details/128569661?spm1001.2101.3001.6650.2utm_mediumdistribute.pc_relevant.none-task-blog-2%7Edefault%7EYuanLiJiHua%7EPosition-2-128569661-blog-124381828.pc_relevant_recovery_v2depth_1-utm_sourcedistribute.pc_relevant.none-task-blog-2%7Edefault%7EYuanLiJiHua%7EPosition-2-128569661-blog-124381828.pc_relevant_recovery_v2utm_relevant_index3使用消息中间件内存中的事件容易丢失我们用MQ层来持久化消息和解耦 Distruptor 框架异步请参考 尼恩的 《Disruptor 框架 红宝书》ForkJoin 框架异步Fork/Join框架是JDK1.7提供的一个并行任务执行框架它可以把一个大任务分成多个可并行执行的子任务然后合并每个子任务的结果得到的大任务的结果。有点类似Hadoop的MapReduceFork/Join框架也可以分成两个核心操作Fork操作将大任务分割成若干个可以并行执行的子任务Join操作合并子任务的执行结果RocketMQ源码中ServiceThread 能急能缓的高性能异步RocketMQ源码中, 实现了一种特殊的高性能异步: 能急能缓 ServiceThread 异步。能急能缓 ServiceThread 异步 有两个特点既能周期性的执行异步任务还能紧急的时候执行应急性的任务RocketMQ的吞吐量达到 70WqpsServiceThread 的异步框架发挥了重要的价值。
