@Async

一、作用：

异步：

异步： 异步调用则是只是发送了调用的指令，调用者无需等待被调用的方法完全执行完毕；而是继续执行下面的流程。例如， 在某个调用中，需要顺序调用 A, B, C三个过程方法；如他们都是同步调用，则需要将他们都顺序执行完毕之后，方算作过程执行完毕； 如B为一个异步的调用方法，则在执行完A之后，调用B，并不等待B完成，而是执行开始调用C，待C执行完毕之后，就意味着这个过程执行完毕了。在Java中，一般在处理类似的场景之时，都是基于创建独立的线程去完成相应的异步调用逻辑，通过主线程和不同的业务子线程之间的执行流程，从而在启动独立的线程之后，主线程继续执行而不会产生停滞等待的情况。

1.1异步调用，通过开启新的线程来执行调用的方法，不影响主线程。异步方法实际的执行交给了Spring的TaskExecutor来完成。

二、使用

2.1 配置类上添加**@EnableAsync**注解

2.2 方法上：

@Async(“自定义一的Executor名称”)

bean的名称不屑，默认是taskExecutor，会默认去该线程池那线程：SimpleAsyncTaskExecutor

@Async 的value 要和2.4中 ThreadPoolTaskExecutor bean的名字一样

2.3 方法返回

\1. 最简单的异步调用，返回值为void
\2. 存在返回值，常调用返回Future

3、有返回值CompletableFuture调用

原理

CompletableFuture 并不使用@Async注解，可达到调用系统线程池处理业务的功能。

JDK5新增了Future接口，用于描述一个异步计算的结果。虽然 Future 以及相关使用方法提供了异步执行任务的能力，但是对于结果的获取却是很不方便，只能通过阻塞或者轮询的方式得到任务的结果。阻塞的方式显然和我们的异步编程的初衷相违背，轮询的方式又会耗费无谓的 CPU 资源，而且也不能及时地得到计算结果。

• CompletionStage代表异步计算过程中的某一个阶段，一个阶段完成以后可能会触发另外一个阶段

• 一个阶段的计算执行可以是一个Function，Consumer或者Runnable。比如：stage.thenApply(x -> square(x)).thenAccept(x -> System.out.print(x)).thenRun(() -> System.out.println())

• 一个阶段的执行可能是被单个阶段的完成触发，也可能是由多个阶段一起触发

  在Java8中，CompletableFuture提供了非常强大的Future的扩展功能，可以帮助我们简化异步编程的复杂性，并且提供了函数式编程的能力，可以通过回调的方式处理计算结果，也提供了转换和组合 CompletableFuture 的方法。

• 它可能代表一个明确完成的Future，也有可能代表一个完成阶段（ CompletionStage ），它支持在计算完成以后触发一些函数或执行某些动作。

• 它实现了Future和CompletionStage接口

private static final ThreadPoolExecutor SELECT_POOL_EXECUTOR = new ThreadPoolExecutor(10, 20, 5000,TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1024), new ThreadFactoryBuilder().setNameFormat("selectThreadPoolExecutor-%d").build());


// tradeMapper.countTradeLog(tradeSearchBean)方法表示，获取数量，返回值为int
// 获取总条数
CompletableFuture<Integer> countFuture = CompletableFuture.supplyAsync(
	() -> tradeMapper.countTradeLog(tradeSearchBean), SELECT_POOL_EXECUTOR);
同步阻塞
CompletableFuture.allOf(countFuture).join();

获取结果
int count = countFuture.get();    

//不需要获取异步结果：
CompletableFuture.runAsync(() -> 	 stockTargetRuleLocalService.addMaxSaleLog(maxSaleLogList));
    public boolean addMaxSaleLog(List<MaxSaleNumLog> maxSaleNumLogList) {
        return maxSaleNumLogMapper.batchInsertReplace(maxSaleNumLogList) > 0;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

三、原理

3.1 @Async注解的就是通过AsyncAnnotationBeanPostProcessor这个后置处理器生成一个代理对象来实现异步的

在这个后置处理器的postProcessAfterInitialization方法中完成了代理

3.2 拦截器invoke方法执行过程：

public Object invoke(final MethodInvocation invocation) throws Throwable {
    Class<?> targetClass = (invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null);
    Method specificMethod = ClassUtils.getMostSpecificMethod(invocation.getMethod(), targetClass);
    final Method userDeclaredMethod = BridgeMethodResolver.findBridgedMethod(specificMethod);
    
    // 异步执行嘛，先获取到一个线程池
    AsyncTaskExecutor executor = determineAsyncExecutor(userDeclaredMethod);
    if (executor == null) {
        throw new IllegalStateException(
            "No executor specified and no default executor set on AsyncExecutionInterceptor either");
    }
    
    // 然后将这个方法封装成一个 Callable对象传入到线程池中执行
    Callable<Object> task = () -> {
        try {
            Object result = invocation.proceed();
            if (result instanceof Future) {
                return ((Future<?>) result).get();
            }
        }
        catch (ExecutionException ex) {
            handleError(ex.getCause(), userDeclaredMethod, invocation.getArguments());
        }
        catch (Throwable ex) {
            handleError(ex, userDeclaredMethod, invocation.getArguments());
        }
        return null;
    };
    // 将任务提交到线程池
    return doSubmit(task, executor, invocation.getMethod().getReturnType());
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

四、踩坑

4.1 默认线程：不指定Executor@Async*(“asyncUpdateMaxSaleQualityExecutor”)*

• 它默认使用的线程池是**SimpleAsyncTaskExecutor：**不是真的线程池，这个类不重用线程，默认每次调用都会创建一个新的线程。

  为每个任务新起一个线程

  默认线程数不做限制

  不复用线程

• 只要你的任务耗时长一点，说不定服务器就给你来个OOM。

• 一般会添加一个线程池的配置，不影响主线程，异步方法交给单独的线程完成

4.2 需要调用异步返回结果:异步方法返回值必须为Future<>，就像Callable与Future。

• 定义异步方法

@Service
public class DeviceProcessServiceImpl implements DeviceProcessService {

    @Autowired
    private DeviceRpcService deviceRpcService;

    @Async("taskExecutor")
    @Override
    public Future<Map<Long, List<ProcessDTO>>> queryDeviceProcessAbilities(List<BindDeviceDO> bindDevices) {
        if (CollectionUtils.isEmpty(bindDevices)) {
            return new AsyncResult<>(Maps.newHashMap());
        }
        List<Long> deviceIds = bindDevices.stream().map(BindDeviceDO::getDeviceId).collect(Collectors.toList());

        List<DeviceInstanceWithProcessResp> devices = deviceRpcService.getDeviceProcessAbility(deviceIds);
        Map<Long, List<ProcessDTO>> deviceAbilityMap = Maps.newHashMap();
        ...
        return new AsyncResult<>(deviceAbilityMap);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

• 调用异步方法，获取结果

  private ProcessAbilityData asyncCollectProcessAbilities(List<BindDeviceDO> bindDevices,
                                                              List<BindStaffDO> bindStaffs, String dccId) {
          // 返回值
          Future<Map<Long, List<ProcessDTO>>> deviceProcessFutureResult = deviceProcessService
              .queryDeviceProcessAbilities(bindDevices);
          Future<Map<String, List<ProcessDTO>>> staffAbilityFutureResult = staffProcessService
              .queryStaffProcessAbilities(bindStaffs, dccId);
          Map<Long, List<ProcessDTO>> deviceAbilityMap;
          Map<String, List<ProcessDTO>> staffAbilityMap;
          try {
              deviceAbilityMap = deviceProcessFutureResult.get();
              staffAbilityMap = staffAbilityFutureResult.get();
          } catch (InterruptedException | ExecutionException e) {
              ...
          }
          return new ProcessAbilityData(deviceAbilityMap, staffAbilityMap);
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16

2、定义异步方法

    @Async
    public Future<String> sayHello1() throws InterruptedException {
        int thinking = 2;
        Thread.sleep(thinking * 1000);//网络连接中 。。。消息发送中。。。
        System.out.println("我爱你啊!");
        return new AsyncResult<String>("发送消息用了"+thinking+"秒");
    }
1
2
3
4
5
6
7

获取一步结果

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration({"classpath:/applicationContext.xml"})
public class TestAsync {
    @Autowired
    private TestAsyncBean testAsyncBean;
    @Test
    public void test_sayHello1() throws InterruptedException, ExecutionException {
        Future<String> future = null;
        System.out.println("你不爱我了么?");
        future = testAsyncBean.sayHello1();
        System.out.println("你竟无话可说, 我们分手吧。。。");
        Thread.sleep(3 * 1000);// 不让主进程过早结束
        System.out.println(future.get());
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

4.3 创建线程池：默认线程池的弊端

线程池不允许使用Executors去创建，不允许使用系统默认的线程池，推荐通过ThreadPoolExecutor的方式，这样的处理方式让开发的工程师更加明确线程池的运行规则，规避资源耗尽的风险。Executors各个方法的弊端：

ExecutorService executorService = Executors.newFixedThreadPool*(*);

• newFixedThreadPool和newSingleThreadExecutor：主要问题是堆积的请求处理队列可能会耗费非常大的内存，甚至OOM。
• newCachedThreadPool和newScheduledThreadPool：要问题是线程数最大数是Integer.MAX_VALUE，可能会创建数量非常多的线程，甚至OOM。

4.4 假如当前类 a.class 中有异步方法，并使用了@Async，那么必须由其他类（例如b.class）来调用，不可由其本身（a.class）来调用

否则aop不生效，类似事物方法调用