Spring Boot 中的异步控制器

Question

我使用 Spring Boot 创建了 Web 服务，其中有一个休息控制器，它通过基于供应商的 JDBC 驱动程序访问数据库并获取记录。在此过程中检索的记录数超过 80K 记录。因此，当我们以 client 的身份访问 rest 端点时，我们会遇到超时错误。

我已尝试使用以下教程设置异步调用。但不幸的是，其余调用仍然超时。

https://howtodoinjava.com/spring-boot2/enableasync-async-controller/

控制器

    @RequestMapping(value = "/v1/lr/fullpositionasync", produces = {APPLICATION_JSON_UTF8_VALUE},   method = RequestMethod.GET)
    @ResponseBody
    public CompletableFuture<List<Position>> retrieveTradePositionsFullAsync(HttpServletRequest request, HttpServletResponse response) throws ExecutionException, InterruptedException {

        CompletableFuture<List<Position>> positionList =null;
        try {
            positionList = positionService.getFullPosition();
        }
        catch(Exception e){
            log.info("Error Occurred in Controller is:"+e.getMessage());
        }
        CompletableFuture.allOf(positionList).join();
        log.info(String.valueOf(positionList.get()));
        return positionList;
    }

服务

@Service
@Slf4j
public class PositionServiceImpl implements PositionService {

    @Autowired
    private PositionDao positionDao;

    @Async("asyncExecutor")
    @Override
    public CompletableFuture<List<Position>> getFullPosition()  {

        List<Position> fullpositionList = null;
        log.info("Getting the full Position process started");
        fullpositionList = positionDao.retrieveData();
        log.info("Total Positions retrieved:"+fullpositionList.size());
        try {
            log.info("Thread is about to sleep 1000 milliseconds");
            Thread.sleep(1000);
        }catch(InterruptedException e){
            log.info(e.getMessage());
        }
        log.info("Full Positions retrieval completed");
        return CompletableFuture.completedFuture(fullpositionList);
    }


}

配置

@Configuration
@EnableAsync
@Slf4j
public class AsyncConfiguration
{
    @Bean(name = "asyncExecutor")
    public Executor asyncExecutor()
    {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(20);
        executor.setMaxPoolSize(1000);
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setThreadNamePrefix("AsynchThreadForEndPoint-");
        executor.initialize();
        log.info("Executor is :"+executor.toString());
        return executor;
    }
}

道

@Repository
public class PositionDaoImpl implements PositionDao {

    @Autowired
    private JdbcTemplate jdbcTemplate;

    private static final String ALL_POSITION_QUERY = "call AllPositionProcedure()";


    public List<Position> retrieveData() {
        return jdbcTemplate.query(ALL_POSITION_QUERY, new BeanPropertyRowMapper(Position.class));
        //  List<Map<String, Object>> mapList = jdbcTemplate.queryForList(sql);

    }

Answer 1

您不能使用 JDBC 对数据库执行异步操作。 JDBC 是阻塞的，所以它会阻塞你的线程，直到操作被执行。如果您想以异步方式执行操作，请使用R2DBC 而不是 JDBC。

Answer 2

对于您的用例，最好的方法是将您的应用程序转换为 Reactive Streams (Flux)。

Flux 是响应式流发布器。它是 JVM 的完全非阻塞反应式编程基础，具有高效的需求管理（以管理“背压”的形式）。它直接与 Java 8 功能 API 集成，特别是 CompletableFuture、Stream 和 Duration。它提供可组合的异步序列 API Flux（用于 [N] 个元素）和 Mono（用于 [0|1] 个元素），广泛实现了Reactive Streams 规范。

在现有应用中实现非常简单。只需更改您的存储库返回类型 Flux 而不是 List 或 Future。

更多信息可以参考Here