異步調用

1.使用：

springboot中的啟動類中需要添加注解@EnableAsync來開啟異步調用，在需要異步執(zhí)行的方法上添加@Async("taskExecutor")注解進行標注。

@Target({ElementType.METHOD, ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface Async {

    /**
     * A qualifier value for the specified asynchronous operation(s).
     * <p>May be used to determine the target executor to be used when executing this
     * method, matching the qualifier value (or the bean name) of a specific
     * {@link java.util.concurrent.Executor Executor} or
     * {@link org.springframework.core.task.TaskExecutor TaskExecutor}
     * bean definition.
     * <p>When specified on a class level {@code @Async} annotation, indicates that the
     * given executor should be used for all methods within the class. Method level use
     * of {@code Async#value} always overrides any value set at the class level.
     * @since 3.1.2
     */
    String value() default "";

}

一般會添加一個線程池的配置，不影響主線程，異步方法交給單獨的線程完成

@Configuration
public class AsyncConfig {

    private static final int MAX_POOL_SIZE = 50;

    private static final int CORE_POOL_SIZE = 20;

    @Bean("taskExecutor")
    public AsyncTaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        taskExecutor.setMaxPoolSize(MAX_POOL_SIZE);
        taskExecutor.setCorePoolSize(CORE_POOL_SIZE);
        taskExecutor.setThreadNamePrefix("async-task-thread-pool");
        taskExecutor.initialize();
        return taskExecutor;
    }
}

需要執(zhí)行異步調用的方法示例(帶有返回值的Future<T>，需要用到AsyncResult)

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

對加了@async注解方法有返回值的調用

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