Java异步编排结合线程池实现多任务按需执行案例

Posted 2021-05-21 北溟溟

前言

在项目中经常会有这样一种场景，在同一个业务中，我们的业务流程会有多个执行步骤，我们最终会把这些业务流程的执行步骤处理结果进行综合处理后返回一个最终结果给前端。按照正常的程序流程串行化执行，可能响应的时间会很长，导致用户体验变差。例如我们在一个业务处理流程中，有5个处理步骤，平均每个处理步骤大概需要1秒钟，那么整个串行化执行过程保守需要5秒钟才能执行完毕，这样加上中间过程处理，可能最终的响应时间远远大于5秒钟，在某些应用场景下，这样的响应效果用户往往是难以接受的。然而真实的情况是我们的整个业务流程处理步骤中，可能只有某些步骤是依赖上一个步骤的处理结果，而其它的步骤是可以并行执行的，串行化执行会导致程序执行过程的阻塞，一步一步执行流程，而并行的方式却可以避免这样的情况，提高程序的执行效率，更快的响应客户的交互需求。在实际的解决方案中我们可以通过闭锁CountDownLatch来解决并行执行的过程，但是CountDownLatch闭锁无法解决部分串行化执行的问题，在java8中，有一种更加合理的多任务处理方案，就是我们的异步编排，可以根据实际执行情况执行我们的程序，不用全部过程阻塞等待。它的核心思想就是短板理论，最终程序的执行时间取决于并行执行业务流程中最后执行完业务的那个执行步骤所花费的时间，如果我们不关心最终的执行结果，那么也可以异步执行直接返回，业务不需要阻塞，从而提高程序执行效率。

正文

• 创建线程池

说明：创建线程的方式主要有四种，分别是①继承Thread类实现②实现Runnable接口③实现Callable接口④Executor 的工具类创建线程池或者手动创建线程池，本案例我们采用第四种方案，根据实际需求创建线程池。

①配置线程池相关配置

atp:
  thread:
    pool:
      #核心线程数量
      corePoolSize: 20
      #最大线程数
      maximumPoolSize: 200
      #空闲线程的最长时间
      keepAliveTime: 10
      #缓存队列大小
      queueSize: 100000

②加载配置文件数据

package com.yundi.atp.platform.config;

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;

/**
 * @Author: yanp
 * @Description: 线程池配置参数
 * @Date: 2021/5/13 11:03
 * @Version: 1.0.0
 */
@Component
@ConfigurationProperties(prefix = "atp.thread.pool")
@Data
public class ThreadPoolConfigProperties {
    /**
     * 核心线程数量
     */
    private Integer corePoolSize;
    /**
     * 最大线程数
     */
    private Integer maximumPoolSize;
    /**
     * 空闲线程的最长时间
     */
    private Integer keepAliveTime;
    /**
     * 队列大小
     */
    private Integer queueSize;
}

③创建线程池实例

package com.yundi.atp.platform.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @Author: yanp
 * @Description: 线程池配置
 * @Date: 2021/5/13 10:38
 * @Version: 1.0.0
 */
@Configuration
public class ThreadPoolConfig {

    @Bean
    public ThreadPoolExecutor threadPoolExecutor(ThreadPoolConfigProperties pool) {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(pool.getCorePoolSize(),
                pool.getMaximumPoolSize(),
                pool.getKeepAliveTime(),
                TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(pool.getQueueSize()),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy());
        return threadPoolExecutor;
    }

}

• 异步编排主要API说明

#执行异步操作，有返回值
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)

#执行异步操作，有返回值，并且可以使用自定义线程池
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)

#执行异步操作，无返回值
public static CompletableFuture<Void> runAsync(Runnable runnable)

#执行异步操作，无返回值，并且可以使用自定义线程池
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)

#执行上一个异步操作后新的异步操作，获取上一个任务返回的结果，并返回当前任务的返回值
public <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn)

#执行上一个异步操作后新的异步操作，获取上一个任务返回的结果，并返回当前任务的返回值，异步操作
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn)

#执行上一个异步操作后新的异步操作，获取上一个任务返回的结果，并返回当前任务的返回值，异步操作加可以自定义线程池
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)

#消费上一个异步操作处理结果。接收上一个异步操作任务的处理结果，并消费处理结果，无返回结果
public CompletableFuture<Void> thenAccept(Consumer<? super T> action)

#消费上一个异步操作处理结果。接收上一个异步操作任务的处理结果，并消费处理结果，无返回结果，异步操作
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action)

#消费上一个异步操作处理结果。接收上一个异步操作任务的处理结果，并消费处理结果，无返回结果，异步操作，异步操作加可以自定义线程池
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action, Executor executor)

#上一个异步操作任务执行后的业务操作
public CompletableFuture<Void> thenRun(Runnable action)

#上一个异步操作任务执行后的业务操作，异步操作
public CompletableFuture<Void> thenRunAsync(Runnable action)

#上一个异步操作任务执行后的业务操作，异步操作，异步操作加可以自定义线程池
public CompletableFuture<Void> thenRunAsync(Runnable action, Executor executor)

#任务组合执行
public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs)

#任意一个任务组合执行完就结束执行
public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs)

• 实现异步编排案例

①控制层

package com.yundi.atp.platform.module.sys.controller;

import com.yundi.atp.platform.common.Result;
import com.yundi.atp.platform.module.sys.UserVo;
import com.yundi.atp.platform.module.sys.service.CompletableFutureService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

/**
 * @Author: yanp
 * @Description: 异步编排任务测试控制层
 * @Date: 2021/5/13 17:26
 * @Version: 1.0.0
 */
@Api(tags = "异步编排业务测试")
@RestController
@RequestMapping(value = "/completableFuture")
public class CompletableFutureController {
    @Autowired
    private CompletableFutureService completableFutureService;

    @ApiOperation(value = "串行查询用户信息详情")
    @PostMapping(value = "/findUserInfoDescBySerial")
    public Result findUserInfoDescBySerial(@RequestParam(value = "userid") String userid) {
        UserVo userVo = completableFutureService.findUserInfoDescBySerial(userid);
        return Result.success(userVo);
    }

    @ApiOperation(value = "并行查询用户信息详情")
    @PostMapping(value = "/findUserInfoDescByParallel")
    public Result findUserInfoDescByParallel(@RequestParam(value = "userid") String userid) {
        UserVo userVo = completableFutureService.findUserInfoDescByParallel(userid);
        return Result.success(userVo);
    }
}

②业务层

package com.yundi.atp.platform.module.sys.service.impl;

import com.yundi.atp.platform.module.sys.UserVo;
import com.yundi.atp.platform.module.sys.entity.User;
import com.yundi.atp.platform.module.sys.service.CompletableFutureService;
import com.yundi.atp.platform.module.sys.service.UserService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @Author: yanp
 * @Description:异步编排测试接口实现类
 * @Date: 2021/5/13 17:41
 * @Version: 1.0.0
 */
@Slf4j
@Service
public class CompletableFutureServiceImpl implements CompletableFutureService {
    @Autowired
    private ThreadPoolExecutor threadPoolExecutor;
    @Autowired
    private UserService userService;

    @Override
    public UserVo findUserInfoDescBySerial(String userid) {
        LocalDateTime startTime = LocalDateTime.now();
        UserVo userVo = new UserVo();
        //1.查询用户信息
        User user = userService.getById(userid);
        BeanUtils.copyProperties(user, userVo);
        //2.业务1结果处理
        String businessOneResult = getBusinessOneResult();
        userVo.setBusinessOne(businessOneResult);
        //3.业务2结果处理
        Integer businessTwoResult = getBusinessTwoResult();
        userVo.setBusinessTwo(businessTwoResult);
        //4.业务3结果处理,基于结果2
        Integer businessThreeResult = getBusinessThreeResult(businessTwoResult);
        userVo.setBusinessThree(businessThreeResult);
        //5.业务4结果处理,基于结果2
        Integer businessFourResult = getBusinessFourResult(businessTwoResult);
        userVo.setBusinessFour(businessFourResult);
        //6.业务5结果处理,基于结果2
        Integer businessFiveResult = getBusinessFiveResult(businessTwoResult);
        userVo.setBusinessFive(businessFiveResult);

        LocalDateTime endTime = LocalDateTime.now();
        log.info("============Serial total execute time:{}s================", ChronoUnit.SECONDS.between(startTime, endTime));
        return userVo;
    }

    @Override
    public UserVo findUserInfoDescByParallel(String userid) {
        LocalDateTime startTime = LocalDateTime.now();
        UserVo userVo = new UserVo();
        //1.查询用户信息
        CompletableFuture<Void> userCompletableFuture = CompletableFuture.runAsync(() -> {
            User user = userService.getById(userid);
            BeanUtils.copyProperties(user, userVo);
        }, threadPoolExecutor);
        //2.业务1结果处理
        CompletableFuture<Void> businessOneCompletableFuture = CompletableFuture.runAsync(() -> {
            String businessOneResult = getBusinessOneResult();
            userVo.setBusinessOne(businessOneResult);
        }, threadPoolExecutor);
        //3.业务2结果处理
        CompletableFuture<Integer> businessTwoCompletableFuture = CompletableFuture.supplyAsync(() -> {
            Integer businessTwoResult = getBusinessTwoResult();
            userVo.setBusinessTwo(businessTwoResult);
            return businessTwoResult;
        }, threadPoolExecutor);
        //4.业务3结果处理,基于结果2
        CompletableFuture<Void> businessThreeCompletableFuture = businessTwoCompletableFuture.thenAcceptAsync(businessTwoResult -> {
            Integer businessThreeResult = getBusinessThreeResult(businessTwoResult);
            userVo.setBusinessThree(businessThreeResult);
        }, threadPoolExecutor);
        //5.业务4结果处理,基于结果2
        CompletableFuture<Void> businessFourResultCompletableFuture = businessTwoCompletableFuture.thenAcceptAsync(businessTwoResult -> {
            Integer businessFourResult = getBusinessFourResult(businessTwoResult);
            userVo.setBusinessFour(businessFourResult);
        }, threadPoolExecutor);
        //6.业务5结果处理,基于结果2
        CompletableFuture<Void> businessFiveResultCompletableFuture = businessTwoCompletableFuture.thenAcceptAsync(businessTwoResult -> {
            Integer businessFiveResult = getBusinessFiveResult(businessTwoResult);
            userVo.setBusinessFive(businessFiveResult);
        }, threadPoolExecutor);
        //7.多任务组合执行
        CompletableFuture.allOf(userCompletableFuture,
                businessOneCompletableFuture,
                businessTwoCompletableFuture,
                businessThreeCompletableFuture,
                businessFourResultCompletableFuture,
                businessFiveResultCompletableFuture).join();
        LocalDateTime endTime = LocalDateTime.now();
        log.info("============Parallel total execute time:{}s================", ChronoUnit.SECONDS.between(startTime, endTime));
        return userVo;
    }

    private Integer getBusinessFiveResult(Integer businessTwoResult) {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return businessTwoResult + 20;
    }

    private Integer getBusinessFourResult(Integer businessTwoResult) {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return businessTwoResult + 10;
    }

    private Integer getBusinessThreeResult(Integer businessTwoResult) {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return businessTwoResult + 5;
    }

    private Integer getBusinessTwoResult() {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return 5;
    }

    private String getBusinessOneResult() {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "businessOneCompletableFuture";
    }
}

• 验证

①串行执行结果：能够正确返回结果，根据打印日志，需要5秒钟响应结果

②并行执行结果：能够正确返回结果，根据打印日志，需要2秒钟响应结果

③通过串行化执行与java异步编排的比较，我们发现异步编排能够极大的提升程序的执行效率，尤其是在我们的多个业务执行逻辑互不干涉的场景中，我们可根据需要对我们的业务做合理的异步编排执行。

结语

ok，到这里我们的异步编排案例实战就结束了，下期见。。。