SpringBoot中的@async正在创建新线程，但控制器在返回响应之前等待async调用完成

Question

上下文

主

@EnableAsync(proxyTargetClass = true)
@EnableScheduling
@SpringBootApplication
public class CardsApplication {

    public static void main(String[] args) {
        SpringApplication.run(CardsApplication.class, args);
    }

    @Bean
    public RestTemplate restTemplate(RestTemplateBuilder builder) {
        return builder.build();
    }

}

控制器

@Validated
@RequiredArgsConstructor
public class CardEligibilityController {
    private final CardEligibilityService cardEligibilityService;

    @PostMapping("/check-eligibility")
    @CrossOrigin(origins = "*")
    public EligibilityResponse checkEligibility(@RequestBody @Valid Applicant applicant){
        return cardEligibilityService.eligibilityService(applicant);
    }
}

服务1

public interface CardEligibilityService {
    EligibilityResponse eligibilityService(Applicant applicant);
}

@Slf4j
@Service
@RequiredArgsConstructor
public class CardEligibilityServiceImpl implements CardEligibilityService {

    private final ThirdPartyEligibilityAdapter thirdPartyEligibilityAdapter;
    private final QueueService queueService;
    private final QueueMessageResponseService queueMessageResponseService;

    @Override
    public EligibilityResponse eligibilityService(Applicant applicant){
        EligibilityResponse eligibilityResponse = checkEligibility(applicant);
        queueService.pushMessage(queueMessageResponseService.createQueueResponse(applicant,eligibilityResponse));
        return eligibilityResponse;
    }

    private EligibilityResponse checkEligibility(Applicant applicant) {
        return thirdPartyEligibilityAdapter.getEligibility(applicant);
    }

}

public interface QueueService {
     void pushMessage(QueueMessage queueMessage);
     void retry();
}


@Service
@RequiredArgsConstructor
@Slf4j
public class QueueServiceImpl implements QueueService{

    private final List<QueueMessage> deadQueue = new LinkedList<>();


    //TODO check why async gets response stuck
    @Override
    @Async
    public void pushMessage(QueueMessage queueMessage){
        try {
            //Push message to a queue - Queue settings Rabbit/Kafka - then this could be
            //used by listeners to persist the data into DB
            log.info("message queued {} ", queueMessage);
        } catch (Exception e) {
            log.error("Error {} , queueMessage {} ", e, queueMessage);
            deadQueue.add(queueMessage);
        }
    }

   
**This method is a fault tolerance mechanism in case push to queue had any issues, The Local Method call to pushMessage isn’t the problem I also tried this by deleting retry method method**

    @Override
    @Scheduled(fixedDelay = 300000)
    public void retry() {
        log.info("Retrying Message push if there are any failure in enqueueing ");
        final List<QueueMessage> temp = new LinkedList<>(deadQueue);
        deadQueue.clear();
        Collections.reverse(temp);
        temp.forEach(this::pushMessage);
    }

}

public interface QueueMessageResponseService {
    QueueMessage createQueueResponse(Applicant applicant, EligibilityResponse eligibilityResponse);
}

@Service
public class QueueMessageResponseServiceServiceImpl implements QueueMessageResponseService {
    @Override
    public QueueMessage createQueueResponse(Applicant applicant, EligibilityResponse eligibilityResponse) {
        return new QueueMessage(applicant,eligibilityResponse);
    }
}

如果我在异步方法中添加thread.sleep(20)；，这将按照预期工作，用户将得到响应，而无需等待async完成。但仍无法了解原因。

@Async
    public void pushMessage(QueueMessage queueMessage) {
        try {
            //Push message to a queue - Queue settings Rabbit/Kafka - then this could be
            //used by listeners to persist the data into DB
            Thread.sleep(20);
            log.info("message queued {} ", queueMessage);
        } catch (Exception e) {
            log.error("Error {} , queueMessage {} ", e, queueMessage);
            deadQueue.add(queueMessage);
        }
    }

Answer 1

对retry中的pushmessage的调用是本地调用。因此不涉及代理,方法是同步执行的。

您必须将async方法移动到它自己的类中。