上海专业网站建设排行,可以做推广的网站有哪些,88hmtopa6c7qq进入,重庆网站优化前言异步执行对于开发者来说并不陌生#xff0c;在实际的开发过程中#xff0c;很多场景多会使用到异步#xff0c;相比同步执行#xff0c;异步可以大大缩短请求链路耗时时间#xff0c;比如#xff1a;「发送短信、邮件、异步更新等」#xff0c;这些都是典型的可以通…前言异步执行对于开发者来说并不陌生在实际的开发过程中很多场景多会使用到异步相比同步执行异步可以大大缩短请求链路耗时时间比如「发送短信、邮件、异步更新等」这些都是典型的可以通过异步实现的场景异步的八种实现方式线程ThreadFuture异步框架CompletableFutureSpring注解AsyncSpring ApplicationEvent事件消息队列第三方异步框架比如Hutool的ThreadUtilGuava异步什么是异步在同步操作中我们执行到 「发送短信」 的时候我们必须等待这个方法彻底执行完才能执行 「赠送积分」 这个操作如果 「赠送积分」 这个动作执行时间较长发送短信需要等待这就是典型的同步场景。实际上发送短信和赠送积分没有任何的依赖关系通过异步我们可以实现赠送积分和发送短信这两个操作能够同时进行比如线程异步public class AsyncThread extends Thread {Overridepublic void run() {System.out.println(Current thread name: Thread.currentThread().getName() Send email success!);}public static void main(String[] args) {AsyncThread asyncThread new AsyncThread();asyncThread.start();}
}当然如果每次都创建一个Thread线程频繁的创建、销毁浪费系统资源我们可以采用线程池private ExecutorService executorService Executors.newCachedThreadPool();public void fun() {executorService.submit(new Runnable() {Overridepublic void run() {log.info(执行业务逻辑...);}});
}可以将业务逻辑封装到Runnable或Callable中交由线程池来执行。Future异步Slf4j
public class FutureManager {public String execute() throws Exception {ExecutorService executor Executors.newFixedThreadPool(1);FutureString future executor.submit(new CallableString() {Overridepublic String call() throws Exception {System.out.println( --- task start --- );Thread.sleep(3000);System.out.println( --- task finish ---);return this is future execute final result!!!;}});//这里需要返回值时会阻塞主线程String result future.get();log.info(Future get result: {}, result);return result;}SneakyThrowspublic static void main(String[] args) {FutureManager manager new FutureManager();manager.execute();}
}输出结果 --- task start --- --- task finish ---Future get result: this is future execute final result!!!CompletableFuture实现异步public class CompletableFutureCompose {/*** thenAccept子任务和父任务公用同一个线程*/SneakyThrowspublic static void thenRunAsync() {CompletableFutureInteger cf1 CompletableFuture.supplyAsync(() - {System.out.println(Thread.currentThread() cf1 do something....);return 1;});CompletableFutureVoid cf2 cf1.thenRunAsync(() - {System.out.println(Thread.currentThread() cf2 do something...);});//等待任务1执行完成System.out.println(cf1结果- cf1.get());//等待任务2执行完成System.out.println(cf2结果- cf2.get());}public static void main(String[] args) {thenRunAsync();}
}我们不需要显式使用ExecutorServiceCompletableFuture 内部使用了ForkJoinPool来处理异步任务如果在某些业务场景我们想自定义自己的异步线程池也是可以的。Spring的Async异步自定义异步线程池/*** 线程池参数配置多个线程池实现线程池隔离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;}
}AsyncServicepublic 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直接实现异步。Spring ApplicationEvent事件实现异步定义事件public class AsyncSendEmailEvent extends ApplicationEvent {/*** 邮箱**/private String email;/*** 主题**/private String subject;/*** 内容**/private String content;/*** 接收者**/private String targetUserId;}定义事件处理器Slf4j
Component
public class AsyncSendEmailEventHandler implements ApplicationListenerAsyncSendEmailEvent {Autowiredprivate IMessageHandler mesageHandler;Async(taskExecutor)Overridepublic void onApplicationEvent(AsyncSendEmailEvent event) {if (event null) {return;}String email event.getEmail();String subject event.getSubject();String content event.getContent();String targetUserId event.getTargetUserId();mesageHandler.sendsendEmailSms(email, subject, content, targerUserId);}
}另外可能有些时候采用ApplicationEvent实现异步的使用当程序出现异常错误的时候需要考虑补偿机制那么这时候可以结合Spring Retry重试来帮助我们避免这种异常造成数据不一致问题。 消息队列回调事件消息生产者Slf4j
Component
public class CallbackProducer {AutowiredAmqpTemplate amqpTemplate;public void sendCallbackMessage(CallbackDTO allbackDTO, final long delayTimes) {log.info(生产者发送消息callbackDTO{}, callbackDTO);amqpTemplate.convertAndSend(CallbackQueueEnum.QUEUE_GENSEE_CALLBACK.getExchange(), CallbackQueueEnum.QUEUE_GENSEE_CALLBACK.getRoutingKey(), JsonMapper.getInstance().toJson(genseeCallbackDTO), new MessagePostProcessor() {Overridepublic Message postProcessMessage(Message message) throws AmqpException {//给消息设置延迟毫秒值通过给消息设置x-delay头来设置消息从交换机发送到队列的延迟时间message.getMessageProperties().setHeader(x-delay, delayTimes);message.getMessageProperties().setCorrelationId(callbackDTO.getSdkId());return message;}});}
} 回调事件消息消费者Slf4j
Component
RabbitListener(queues message.callback, containerFactory rabbitListenerContainerFactory)
public class CallbackConsumer {Autowiredprivate IGlobalUserService globalUserService;RabbitHandlerpublic void handle(String json, Channel channel, Headers MapString, Object map) throws Exception {if (map.get(error) ! null) {//否认消息channel.basicNack((Long) map.get(AmqpHeaders.DELIVERY_TAG), false, true);return;}try {CallbackDTO callbackDTO JsonMapper.getInstance().fromJson(json, CallbackDTO.class);//执行业务逻辑globalUserService.execute(callbackDTO);//消息消息成功手动确认对应消息确认模式acknowledge-mode: manualchannel.basicAck((Long) map.get(AmqpHeaders.DELIVERY_TAG), false);} catch (Exception e) {log.error(回调失败 - {}, e);}}
}ThreadUtil异步工具类Slf4j
public class ThreadUtils {public static void main(String[] args) {for (int i 0; i 3; i) {ThreadUtil.execAsync(() - {ThreadLocalRandom threadLocalRandom ThreadLocalRandom.current();int number threadLocalRandom.nextInt(20) 1;System.out.println(number);});log.info(当前第 i 个线程);}log.info(task finish!);}
}Guava异步Guava的ListenableFuture顾名思义就是可以监听的Future是对java原生Future的扩展增强。我们知道Future表示一个异步计算任务当任务完成时可以得到计算结果。如果我们希望一旦计算完成就拿到结果展示给用户或者做另外的计算就必须使用另一个线程不断的查询计算状态。这样做代码复杂而且效率低下。使用「Guava ListenableFuture」可以帮我们检测Future是否完成了不需要再通过get()方法苦苦等待异步的计算结果如果完成就自动调用回调函数这样可以减少并发程序的复杂度。ListenableFuture是一个接口它从jdk的Future接口继承添加了void addListener(Runnable listener, Executor executor)方法。我们看下如何使用ListenableFuture。首先需要定义ListenableFuture的实例:Guava的ListenableFuture顾名思义就是可以监听的Future是对java原生Future的扩展增强。我们知道Future表示一个异步计算任务当任务完成时可以得到计算结果。如果我们希望一旦计算完成就拿到结果展示给用户或者做另外的计算就必须使用另一个线程不断的查询计算状态。这样做代码复杂而且效率低下。使用「Guava ListenableFuture」可以帮我们检测Future是否完成了不需要再通过get()方法苦苦等待异步的计算结果如果完成就自动调用回调函数这样可以减少并发程序的复杂度。
ListenableFuture是一个接口它从jdk的Future接口继承添加了void addListener(Runnable listener, Executor executor)方法。
我们看下如何使用ListenableFuture。首先需要定义ListenableFuture的实例:首先通过MoreExecutors类的静态方法listeningDecorator方法初始化一个ListeningExecutorService的方法然后使用此实例的submit方法即可初始化ListenableFuture对象。ListenableFuture要做的工作在Callable接口的实现类中定义这里只是休眠了1秒钟然后返回一个数字1有了ListenableFuture实例可以执行此Future并执行Future完成之后的回调函数。 Futures.addCallback(listenableFuture, new FutureCallbackInteger() {Overridepublic void onSuccess(Integer result) {//成功执行...System.out.println(Get listenable futures result with callback result);}Overridepublic void onFailure(Throwable t) {//异常情况处理...t.printStackTrace();}
});
