实现异步的8种方式

2023-03-28 01 Java java 0 评论阅读时长: 12(分)

1. 前言

异步执行对于开发者来说并不陌生，在实际的开发过程中，很多场景多会使用到异步，相比同步执行，异步可以大大缩短请求链路耗时时间，比如：「发送短信、邮件、异步更新等」，这些都是典型的可以通过异步实现的场景

首先我们先看一个常见的用户下单的场景：

在同步操作中，我们执行到 「发送短信」 的时候，我们必须等待 「赠送积分」 这个操作彻底执行完才能执行，如果 「赠送积分」 这个动作执行时间较长，发送短信需要等待，这就是典型的同步场景。

实际上，发送短信和赠送积分没有任何的依赖关系，通过异步，我们可以实现赠送积分和发送短信这两个操作能够同时进行，比如

2. 异步的8种实现方式

线程Thread
Future
异步框架CompletableFuture
Spring注解@Async
Spring ApplicationEvent事件
消息队列
第三方异步框架(比如 Hutool的ThreadUtil)
Guava异步

3. 异步案例代码

3.1 线程异步

/**
 * @author xiaoyuge
 */
public class AsyncThread extends Thread{

    @Override
    public void run() {
        System.out.println("current thread name:"+Thread.currentThread().getName());
    }

    public static void main(String[] args) {
        AsyncThread asyncThread = new AsyncThread();
        asyncThread.start();
    }
}

当然如果每次都创建一个Thread线程，频繁的创建、销毁，浪费系统资源，可以采用线程池：

public class AsyncThreadPool {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newCachedThreadPool();

        executorService.submit(() -> System.out.println("执行业务逻辑......"));
    }
}

可以将业务逻辑封装到Runnable或Callable中，交由线程池来执行。

3.2 Future异步

/**
 * @author xiaoyuge
 */
public class FutureDemo {

    public String executeDemo() throws ExecutionException, InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(1);
        Future<String> future = executorService.submit(() -> {
            System.out.println(" --- task start --- ");
            Thread.sleep(3000);
            System.out.println(" --- task finish ---");
            return "this is future execute final result!!!";
        });
        //这里需要返回值时会阻塞主线程
        return future.get();
    }

    public static void main(String[] args) {
        FutureDemo demo = new FutureDemo();
        try {
            System.out.println(demo.executeDemo());
        } catch (ExecutionException | InterruptedException e) {
            e.printStackTrace();
        }
    }
}

输出结果：

1
2
3

 --- task start --- 
 --- task finish ---
this is future execute final result!!!

Future的不足之处:

无法被动接收异步任务的计算结果： 虽然我们可以主动将异步任务提交给线程池中的线程来执行，但是待异步任务执行结束之后，主线程无法得到任务完成与否的通知，它需要通过get方法主动获取任务执行的结果

Future间彼此孤立： 有时某一个耗时很长的异步任务执行结束之后，你想利用它返回的结果再做进一步的运算，该运算也会是一个异步任务，两者之间的关系需要程序开发人员手动进行绑定赋予，Future并不能将其形成一个任务流（pipeline），每一个Future都是彼此之间都是孤立的，所以才有了后面的CompletableFuture，CompletableFuture就可以将多个Future串联起来形成任务流。

Future没有很好的错误处理机制： 截止目前，如果某个异步任务在执行发的过程中发生了异常，调用者无法被动感知，必须通过捕获get方法的异常才知晓异步任务执行是否出现了错误，从而在做进一步的判断处理

3.3 CompletableFuture实现异步

/**
 * @author xiaoyuge
 */
public class CompletableFutureDemo {

    public static void thenRunAsync() throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> cf1 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + " cf1 do something");
            return 1;
        });

        CompletableFuture<Integer> cf2 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + " cf2 do something");
            return 2;
        });
        System.out.println("cf1执行结果 ->" + cf1.get());
        System.out.println("cf2执行结果 ->" + cf2.get());
    }

    public static void main(String[] args) {
        try {
            thenRunAsync();
        } catch (ExecutionException | InterruptedException e) {
            e.printStackTrace();
        }
    }
}

输出结果：

ForkJoinPool.commonPool-worker-1 cf1 do something
ForkJoinPool.commonPool-worker-1 cf2 do something
cf1执行结果 ->1
cf2执行结果 ->2

我们不需要显式使用ExecutorService，CompletableFuture 内部使用了ForkJoinPool来处理异步任务，如果在某些业务场景我们想自定义自己的异步线程池也是可以的。

3.4 Spring的@Async异步

自定义线程池

@EnableAsync        //异步调用@Async注解生效。
@Configuration
public class TaskPoolConfig {

    @Bean("taskExcutor")
    public Executor takExecutor(){
        //返回可用处理器的Java虚拟机的数量
        int i = Runtime.getRuntime().availableProcessors();
        System.out.println("系统最大线程数："+i);
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(16);   //核心线程池大小
        executor.setMaxPoolSize(20); //最大线程数
        executor.setKeepAliveSeconds(60);//活跃时间
        executor.setQueueCapacity(9999); //配置缓存队列容量，默认值为：Integer.MAX_VALUE
        executor.setThreadNamePrefix("async Excutor -");//线程名前缀
        executor.setAwaitTerminationSeconds(60);  //等待的时间
        //调度器shutdown被调用时等待当前被调度的任务完成, 就是等待所有的任务结束后再关闭线程池
        executor.setWaitForTasksToCompleteOnShutdown(true);
        // 线程池对拒绝任务（无线程可用）的处理策略，目前只支持AbortPolicy、CallerRunsPolicy；默认为后者
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }
}

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 {

    @Autowired
    private IMessageHandler mesageHandler;

    @Override
    @Async("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);
        }
    }
    
    @Override
    @Async("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实现异步

3.5 Spring ApplicationEvent事件实现异步

定义事件

public class AsyncSendEmailEvent extends ApplicationEvent {
    /**
     * 邮箱
     **/
    private String email;
   /**
     * 主题
     **/
    private String subject;
    /**
     * 内容
     **/
    private String content;
  
    /**
     * 接收者
     **/
    private String targetUserId;
   
    //------省略getter/setter方法
}

定义事件处理器

@Slf4j
@Component
public class AsyncSendEmailEventHandler implements ApplicationListener<AsyncSendEmailEvent> {

    @Autowired
    private IMessageHandler messageHandler;
    
    @Async("taskExecutor")
    @Override
    public void onApplicationEvent(AsyncSendEmailEvent event) {
        if (event == null) {
            return;
        }
        String email = event.getEmail();
        String subject = event.getSubject();
        String content = event.getContent();
        String targetUserId = event.getTargetUserId();
        messageHandler.sendsendEmailSms(email, subject, content, targerUserId);
      }
}

另外，可能有些时候采用ApplicationEvent实现异步的使用，当程序出现异常错误的时候，需要考虑补偿机制，那么这时候可以结合Spring Retry重试来帮助我们避免这种异常造成数据不一致问题

3.6 消息队列

配置

spring:
  rabbitmq:
    ####连接地址
    host: 192.168.0.92
    ####端口号
    port: 5672
    ####账号
    username: xiaoyuge
    ####密码
    password: 123456
    ### 地址
    virtual-host: /mytest1205

@Component
public class RabbitMQConfig {
    /**
     * 定义交换机
     */
    private String EXCHANGE_SPRINGBOOT_NAME = "springboot_topic_exchange";
    /**
     * 短信队列
     */
    private String FANOUT_SMS_QUEUE = "springboot_topic_sms_queue";
    /**
     * 邮件队列
     */
    private String FANOUT_SMS_EMAIL = "springboot_topic_email_queue";

    /**
     * 创建短信队列
     */
    @Bean
    public Queue smsQueue() {
        return new Queue(FANOUT_SMS_QUEUE);
    }

    /**
     * 创建邮件队列
     */
    @Bean
    public Queue emailQueue() {
        return new Queue(FANOUT_SMS_EMAIL);
    }

    /**
     * 创建交换机
     *
     * @return
     */
    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange(EXCHANGE_SPRINGBOOT_NAME);
    }

    /**
     * 定义短信队列绑定交换机
     */
    @Bean
    public Binding smsBindingExchange(Queue smsQueue, TopicExchange topicExchange) {
        return BindingBuilder.bind(smsQueue).to(topicExchange).with("my.sms");
    }
    /**
     * 定义邮件队列绑定交换机
     */
    @Bean
    public Binding emailBindingExchange(Queue emailQueue, TopicExchange topicExchange) {
        return BindingBuilder.bind(emailQueue).to(topicExchange).with("my.#");
    }
}

回调事件消息生产者

//<dependency>
//   <groupId>org.springframework.boot</groupId>
//   <artifactId>spring-boot-starter-amqp</artifactId>
//</dependency>
@Slf4j
@Component
public class CallbackProducer {

    @Autowired
    private AmqpTemplate amqpTemplate;
    public void send(CallbackDTO callbackDTO, final long delay){
        log.info("生产者发送消息，callbackDTO,{}", callbackDTO);
        // 参数1:交换机名称 、参数2: 路由key  参数3:消息
        amqpTemplate.convertAndSend("springboot_topic_exchange","my.time","this is a message");
    }
}

回调事件消息消费者

@Component
@RabbitListener(queues = "springboot_topic_email_queue")
public class CallbackConsumer {

    @RabbitHandler
    public void consumer(String mesg){
        System.out.println("邮件消费者接收到消息:"+mesg);
    }
}
@Component
@RabbitListener(queues = "springboot_topic_sms_queue")
class FanoutSmsConsumer {

    @RabbitHandler
    public void consumer(String msg){
        System.out.println("短信消费者收到消息："+msg);
    }
}

3.7 ThreadUtil异步工具类

ThreadUtil.execAsync

@Slf4j
public class ThreadUtils {
   /**
    * <dependency>
    *     <groupId>cn.hutool</groupId>
    *     <artifactId>hutool-all</artifactId>
    *     <version>5.7.4</version>
    * </dependency>
    */
    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);
            });
           System.out.println("当前第：" + i + "个线程");
        }
       System.out.println("task finish!");
    }
}

输出结果：

当前第：0个线程
当前第：1个线程
当前第：2个线程
task finish!
3
11
10

ThreadUtil.execute

public static void main(String[] args) throws InterruptedException {
        CountDownLatch countDownLatch = ThreadUtil.newCountDownLatch(5);
        for (int i = 0; i < countDownLatch.getCount(); i++) {
            ThreadUtil.execute(()->{
                try {
                    Thread.sleep(2000); //休眠2秒
                    ThreadLocalRandom threadLocalRandom = ThreadLocalRandom.current();
                    int number = threadLocalRandom.nextInt(20) + 1;
                    System.out.println(number);
                }catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
            System.out.println("第"+ i +"个线程" );
            //调用线程计数器 - 1
            countDownLatch.countDown();
        }
        //唤醒主线程
        countDownLatch.await();
        System.out.println(" finish ");
    }
}

输出结果：

第0个线程
第1个线程
第2个线程
//这里等了2s
18
13
5

3.8 Guava异步

Guava的ListenableFuture顾名思义就是可以监听的Future，是对java原生Future的扩展增强。我们知道Future表示一个异步计算任务，当任务完成时可以得到计算结果。如果我们希望一旦计算完成就拿到结果展示给用户或者做另外的计算，就必须使用另一个线程不断的查询计算状态。这样做，代码复杂，而且效率低下。使用「Guava ListenableFuture」可以帮我们检测Future是否完成了，不需要再通过get()方法苦苦等待异步的计算结果，如果完成就自动调用回调函数，这样可以减少并发程序的复杂度

ListenableFuture是一个接口，它从jdk的Future接口继承，添加了void addListener(Runnable listener, Executor executor)方法。

看下如何使用ListenableFuture。首先需要定义ListenableFuture的实例:

ListeningExecutorService executorService = MoreExecutors.listeningDecorator(Executors.newCachedThreadPool());
        final ListenableFuture<Integer> listenableFuture = executorService.submit(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                log.info("callable execute...")
                TimeUnit.SECONDS.sleep(1);
                return 1;
            }
        });

首先通过MoreExecutors类的静态方法listeningDecorator方法初始化一个ListeningExecutorService的方法，然后使用此实例的submit方法即可初始化ListenableFuture对象。

ListenableFuture要做的工作，在Callable接口的实现类中定义，这里只是休眠了1秒钟然后返回一个数字1，有了ListenableFuture实例，可以执行此Future并执行Future完成之后的回调函数。

Futures.addCallback(listenableFuture, new FutureCallback<Integer>() {
    @Override
    public void onSuccess(Integer result) {
        //成功执行...
        System.out.println("Get listenable future's result with callback " + result);
    }

    @Override
    public void onFailure(Throwable t) {
        //异常情况处理...
        t.printStackTrace();
    }
});

本博文摘录自：austin流川枫的实现异步的8种方式

本文链接： http://xiaoyuge.work/async/index.html

版权声明： 本博客所有文章除特别声明外，均采用 CC BY 4.0 CN协议许可协议。转载请注明出处！本博客如有侵权，请联系删除！

小余哥xiaoyouge

分享一些技术教程与实践，以及日常踩坑记录。