异步

什么是异步

异步编程是一种编程范式，它允许程序在等待操作完成（如等待网络响应、文件读写等）时继续执行其他任务。这种编程方式对于提高程序的性能和响应性至关重要，尤其是在处理耗时操作或在资源受限的环境中。下面我将更详细地探讨异步编程的原理、实现方式、应用场景以及它带来的挑战。

想象一下你正在家里做饭。如果你使用同步的方式来做菜，你会先做第一个菜，等它完全做好并摆上桌后，才开始做第二个菜。这种方式非常低效，因为你必须等待每个菜完全做好才能开始下一个，这会导致整个做饭过程非常漫长。

如果你使用异步的方式，你会同时准备多个菜。当你把第一个菜放在炉子上煮时，你可以开始切第二个菜的食材，然后当你等待第一个菜煮熟的时候，你可以开始烹饪第二个菜。这样，你可以同时进行多个任务，而不需要等待一个任务完全完成才开始下一个。这种方式更加高效，因为你最大化了你的时间和资源利用率，最终可以更快地完成所有菜肴。

异步编程的原理

在传统的同步编程模型中，程序会顺序执行，一个任务必须等待前一个任务完成后才能开始执行。这种方式在处理耗时操作时会导致程序界面冻结或响应变慢，因为CPU在等待操作完成期间无法执行其他任务。

异步编程通过将耗时任务与主程序流程分离来解决这个问题。当一个异步任务被触发时，它会在后台运行，而主程序流程可以继续执行其他操作。当异步任务完成时，它会通知主程序流程，然后主程序流程可以处理任务的结果。

实现异步编程的方式

异步编程可以通过多种机制实现，包括：

• 回调函数：在触发异步操作时提供一个回调函数，当操作完成时调用这个函数。这种方式在JavaScript中非常常见，但由于可能导致“回调地狱”（深层嵌套的回调函数）。
• 事件驱动：在这种模型中，异步操作会触发事件，而事件监听器会响应这些事件。这种方式在Node.js中广泛使用。
• Promises：Promises是现代JavaScript中处理异步操作的一种方式，它提供了一种链式调用的方式来处理异步逻辑，避免了回调地狱的问题。
• 异步/await：这是Promises的语法糖，它允许开发者以同步代码的方式编写异步逻辑，使得代码更加清晰易读。

应用场景

异步编程在多种场景中都非常有用，包括但不限于：

• 网络请求：在等待网络响应时，程序可以继续执行其他任务，而不是冻结等待。
• 文件操作：读写文件通常是耗时的，使用异步编程可以避免在文件操作完成前阻塞程序。
• 数据库操作：与文件操作类似，数据库操作也可以异步执行，以提高程序的效率。
• 用户界面：确保应用程序界面即使在后台进行复杂计算或数据加载时也能保持响应。

异步编程的挑战

尽管异步编程带来了许多好处，但它也引入了一些挑战：

• 复杂性：异步代码的执行流程可能比同步代码更难理解和调试，因为它涉及到并行执行和时间依赖性。
• 状态管理：在异步编程中，跟踪程序状态可能会变得复杂，因为多个任务可能同时进行。
• 错误处理：错误处理在异步编程中可能更加复杂，因为错误可能在任何时候发生，并且可能不在主程序流程中。
• 资源竞争：在多线程环境中，异步操作可能会导致资源竞争，需要仔细设计同步机制来避免死锁和竞态条件。

8 种异步实现方式归纳总结

一、异步的八种实现方式

1. 线程Thread
2. Future
3. 异步框架CompletableFuture
4. [Spring注解] @Async
5. Spring ApplicationEvent事件
6. 消息队列
7. 第三方异步框架，比如Hutool的ThreadUtil
8. Guava异步

二、什么是异步？

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

什么是异步？

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

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

三、异步编程

1、线程异步

public class AsyncThread extends Thread {
    @Override
    public 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.run();
    }
}

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

private ExecutorService executorService = Executors.newCachedThreadPool();

public void fun() {
    executorService.submit(new Runnable() {
        @Override
        public void run() {
            log.info("执行业务逻辑...");
        }
    });
}

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

2、 Future异步

@Slf4j
public class FutureManager {
    public String execute() throws Exception {
        ExecutorService executor = Executors.newFixedThreadPool(1);
        Future<String> future = executor.submit(new Callable<String>() {
            @Override
            public 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;
    }

    @SneakyThrows
    public static void main(String[] args) {
        FutureManager manager = new FutureManager();
        manager.execute();
    }
}

输出结果：

--- task start --- 
 --- task finish ---
 Future get result: this is future execute final result!!!

（1） Future的不足之处

Future的不足之处的包括以下几点：

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

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

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

3、CompletableFuture实现异步

public class CompletableFutureCompose {
    /**
     * thenAccept子任务和父任务公用同一个线程
     */
    @SneakyThrows
    public static void thenRunAsync() {
        CompletableFuture<Integer> cf1 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread() + " cf1 do something....");
            return 1;
        });
        CompletableFuture<Void> 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();
    }
}

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

4、Spring的@Async异步

（1）自定义异步线程池

/**
 * 线程池参数配置，多个线程池实现线程池隔离，@Async注解，默认使用系统自定义线程池，可在项目中设置多个线程池，在异步调用的时候，指明需要调用的线程池名称，比如：@Async("taskName")
 **/
@EnableAsync
@Configuration
public class TaskPoolConfig {
    /**
     * 自定义线程池
     **/
    @Bean("taskExecutor")
    public Executor taskExecutor() {
        //返回可用处理器的Java虚拟机的数量 12
        int i = Runtime.getRuntime().availableProcessors();
        System.out.println("系统最大线程数  ：" + i);
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        //核心线程池大小
        executor.setCorePoolSize(16);
        //最大线程数
        executor.setMaxPoolSize(20);
        //配置队列容量，默认值为Integer.MAX_VALUE
        executor.setQueueCapacity(99999);
        //活跃时间
        executor.setKeepAliveSeconds(60);
        //线程名字前缀
        executor.setThreadNamePrefix("asyncServiceExecutor -");
        //设置此执行程序应该在关闭时阻止的最大秒数，以便在容器的其余部分继续关闭之前等待剩余的任务完成他们的执行
        executor.setAwaitTerminationSeconds(60);
        //等待所有的任务结束后再关闭线程池
        executor.setWaitForTasksToCompleteOnShutdown(true);
        return executor;
    }
}

（2） 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直接实现异步。

5、Spring ApplicationEvent事件实现异步

（1）定义事件

public class AsyncSendEmailEvent extends ApplicationEvent {
    /**
     * 邮箱
     **/
    private String email;
   /**
     * 主题
     **/
    private String subject;
    /**
     * 内容
     **/
    private String content;
  
    /**
     * 接收者
     **/
    private String targetUserId;

}

（2）定义事件处理器

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

    @Autowired
    private IMessageHandler mesageHandler;
    
    @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();
        mesageHandler.sendsendEmailSms(email, subject, content, targerUserId);
      }
}

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

6、消息队列

（1）回调事件消息生产者

@Slf4j
@Component
public class CallbackProducer {

    @Autowired
    AmqpTemplate 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() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                //给消息设置延迟毫秒值，通过给消息设置x-delay头来设置消息从交换机发送到队列的延迟时间
                message.getMessageProperties().setHeader("x-delay", delayTimes);
                message.getMessageProperties().setCorrelationId(callbackDTO.getSdkId());
                return message;
            }
        });
    }
}

（2）回调事件消息消费者

@Slf4j
@Component
@RabbitListener(queues = "message.callback", containerFactory = "rabbitListenerContainerFactory")
public class CallbackConsumer {

    @Autowired
    private IGlobalUserService globalUserService;

    @RabbitHandler
    public void handle(String json, Channel channel, @Headers Map<String, 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: manual
            channel.basicAck((Long) map.get(AmqpHeaders.DELIVERY_TAG), false);
        } catch (Exception e) {
            log.error("回调失败 -> {}", e);
        }
    }
}

7、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!");
    }
}

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();
    }
});

消息队列（MQ）

1、什么是消息队列

消息队列：一般我们会简称它为MQ(Message Queue)。
Message Query（MQ），消息队列中间件，很多初学者认为，MQ通过消息的发送和接受来实现程序的异步和解耦，mq主要用于异步操作，这个不是mq的真正目的，只不过是mq的应用，mq真正的目的是为了通讯。
他屏蔽了复杂的通讯协议，像常用的dubbo，http协议都是同步的。
这两种协议很难实现双端通讯，A调用B,B也可以主动调用A，而且不支持长连接。mq做的就是在这些协议上构建一个简单协议——生产者、消费者模型，mq带给我们的不是底层的通讯协议，而是更高层次的通讯模型。他定义了两个对象：发送数据的叫做生产者，接受消息的叫做消费者，我们可以无视底层的通讯协议，我们可以自己定义生产者消费者。

2、使用场景

最主要的作用是异步、削峰、解耦

3、消息队列与传统设计的区别

1、传统设计

这是一个用户购物下单的流程图：

观察一下上面的设计，属于典型的串行化调用，这种设计模式有一个很大的优势，就是代码简单，出现问题很容易定位到问题。但是也有很多劣势，下面咱们从三高（高并发，高性能，高可用）三个方面去评审一下这个设计。

高并发：因为这些操作都是由一个线程（主线程）去执行这些操作，所以当我们的QPS如果很高的话，很容易造成超时。
高性能：因为上面这种设计模式是串行的，假设我的每次网络传输耗时200ms，业务处理需要20ms，完成上面那些操作需要耗时2s，这样用户体验也会很差（想象一下每次下单都需要等2s）,如果用户下单后的操作越来越多，耗时只会越来越高。
所以在一个大型的互联网项目中，以上设计是完全不可取的（非核心模块除外）。
高可用：这些服务假如有一个服务挂掉（宕机或者网络波动），就意味着我这个请求失败了，这样用户体验会极差，用户会频繁看到支付失败。

2、并行处理调优

既然上面说的是串行模块，那么我们用自己的线程池把他改为并行的设计，再看评审一下。
所谓的并行设计就是原来由一个线程去串行做的逻辑，改为多个线程并行去做。

高可用：这些服务假如有一个服务挂掉（宕机或者网络波动），理论上讲，如果补偿服务做的出色的话，还是满足高可用的。（可以用try,catch）
高并发：相比上面的设计，系统的吞吐量可以达到了很大程度上的提升。
高性能：相比上面的设计，因为很多业务是并行执行的，所以相当于只有200*2+20，就可以返回。
上面这个设计看起来还是不错的设计，所以在很多这种串行调用，多次io的时候，我们就可以采用这种方案，上面这种设计也是多线程的一种实战应用。
下面来分析一下弊端：
1.系统的可扩展性太差了。上面只是列举了4步，但是实际上会有几十步，这几十步放到代码里就会像屎堆一样，可维护性极差。每次加一个步骤，都要多调一个接口，然后重新发布一下服务。
2，系统的耦合性太高了。想象一下，几十个http调用放到一起并发执行，很有可能会影响其他的点，尤其是淘宝京东这种秒杀敏感的业务，和钱挂钩的业务，很容易出现p0级别的bug。
3，使用的业务本身的线程池，在并发很多的情况下，容易造成cpu的竞争。

3、消息队列

咱们继续从三高的层面去审视一下这个设计：

高可用：当我系统里的一个模块宕机了，不会影响到我其他服务。（可以通过数据补偿或者分布式事务来保证数据最终一致性）
高性能：用户下单，将下单所需要的数据都放到消息队列里，就直接返回了，所有耗时相当于就是网络传输所耗时。
高并发：由于消息队列不处理任何业务上的逻辑，所以它支持的并发是百万级别的。假如有100万个用户下单，100万的数据放到消息队列里，连接消息队列的服务慢慢消费即可，也不至于造成瞬间有百万请求进来，将我的服务压垮。

4、三大优点

1、异步

连接消息队列的服务可以异步去执行。而且每次多增加一个步骤，我下单的代码是不需要动的，只需要再增加一个消费者即可。

2、削峰

比如说我平时服务就只能支撑几万的qps，像淘宝京东那种秒杀，那时候服务突然打进来（如果采用第二种方案）那服务就会直接被压死了。但是如果采用消息队列，这秒杀进来的所有的请求都不会直接打到具体服务上，都会先打到消息队列里，然后我后面的服务再慢慢消费。
可以看看淘宝京东双11秒杀的时候，是不是有的时候慢是慢了点，但是服务起码没挂。等我秒杀结束之后，服务还能正常运转。

消息队列就像是一个三峡大坝，用来拦截上游给的压力。

3、解耦

就像高可用里面说的一样，发淘金币服务挂了，和下单有什么关系，发淘金币服务挂了，我还是可以正常下单，只不过后期可以数据补偿或者分布式事务去解决这个问题

5、缺点

1、增加了系统复杂性

所以说如果说你的业务量不大，并发也不高，就没必要使用消息队列。

2、事务问题

事务问题其实是分布式系统肯定会存在的一个问题，只不过消息队列更严重一些。一般解决方案有两种，第一种就是采用分布式事务，这个下单的里涉及的所有服务放到一个事务里面，要么都成功，要么都失败。第二种就是，消费者做好合理的数据补偿措施，比如说，消息重试，人工刷数据等等。

3、可用性

刚才讲了解耦，其实是系统的各个模块之间的解耦，但是这些模块都和消息队列关联，万一消息队列挂了，就真的下不了单了。为了保证可用性，我们可以采用消息队列集群，前端流量限流等，后面会介绍。

6、MQ常见问题

1、消息堆积问题怎么解决

什么是消息堆积：生产者的生产速率远远大于消费者的消费速率，使消息大批量的堆积在消息队列。
解决方案：

提升消费者的消费速率（增加消费者集群）
消费者分批多线程去处理
限流，保证进入到消息队列的都是有用的消息

2、重复消费问题怎么解决

产生原因：

生产者产生了两条一模一样的消息
消费者一条消息消费了多遍
消息重试：消息重试一般发生于一个消费者发生了异常（网络波动或者系统假死），这个时候这个消费者就会通知生产者重新发送。就会带来重复消费的问题。
解决方法：
幂等性校验
可以采用常用的幂等解决方案（分布式锁），全局id+业务场景保证唯一性。处理消息的时候，先看一下缓存中有没有，如果有的话，就认为你已经被消费了，默认处理完成，直接return就好了。所有的重复提交问题，都可以用幂等性来解决。为了保险起见，也可以在数据库上做好唯一索引。

3、如果避免消息丢失

消息确认机制，生产者必须确认消息成功刷盘到硬盘中，才确认消息发送成功。
消息持久化机制，作为mq中间件，会把消息持久化到硬盘，因为内存里的数据断电就丢失了
消费者必须确认消息消费成功，否则进行重试，重试达到一定次数后，通知开发人员做好补偿措施
需要关闭自动提交，当消费者消费的完成后，再进行手动提交，通知mq我消费成功了，防止mq的自动提交。
自动提交就是当消费者接收到消息后，mq过了几秒自动认为消费者已经消费了。但这个时候消费者如果挂了，这条消息就没有被消费。

4、如何做到顺序消费

大多数的项目不需要保证顺序一致性，某些特殊场景必须保证顺序一致性，比如说，mq用于保证redis和mysql的数据一致性。
绑定同一个消费队列，消费的时候要注意如果使用了多线程处理，避免重新创建list，要在原来的list进行修改。

5、消费者怎么知道mq里有消息了

两种方案：
1，mq主动通知（push）
当mq中有消息，就会通知消费者来进行消费。这种模型有一个致命伤，就是慢消费。
2，消费者轮询（pull）
消费者去轮询看看有没有自己要消费的消息。这种模型也有弊端就是消息延迟与忙等。
先来看push
如果消费者的速度比发送者的速度慢很多，势必造成消息在mq的堆积。假设这些消息都是有用的无法丢弃的，消息就要一直在mq端保存。当然这还不是最致命的，最致命的是mq给消费者推送一堆无法处理的消息，消费者不是拒绝就是报错，然后来回踢皮球。
反观pull模式，消费者可以按需消费，不用担心自己处理不了的消息来骚扰自己，而mq堆积消息也会相对简单，无需记录每一个要发送消息的状态，只需要维护所有消息的队列和偏移量就可以了。所以消息量有限且到来的速度不均匀的情况，pull模式比较合适。
由于主动权在消费方，消费方无法准确地决定何时去拉取最新的消息。如果一次pull取到消息了还可以继续去pull，如果没有pull取到则需要等待一段时间重新pull。
但等待多久就很难判定了。当然也不是说延迟就没有解决方案了，业界较成熟的做法是从短时间开始（不会对mq有太大负担），然后指数级增长等待。比如开始等5ms，然后10ms，然后20ms，然后40ms……直到有消息到来，然后再回到5ms。
即使这样，依然存在延迟问题：假设40ms到80ms之间的50ms消息到来，消息就延迟了30ms，而且对于半个小时来一次的消息，这些开销就是白白浪费的。
在阿里的RocketMq里，有一种优化的做法-长轮询，来平衡推拉模型各自的缺点。基本思路是:消费者如果尝试拉取失败，不是直接返回,而是把连接挂在那里等待,服务端如果有新的消息到来，把连接复用起来，这也是不错的思路。但海量的长连接mq对系统的开销还是不容小觑的，还是要合理的评估时间间隔。

7、几种常见的消息队列