Spring Boot @Async Controller with CompletableFuture

Learn to create asynchronous methods in the Spring framework with the help of @Async and @EnableAsync annotations that use a thread pool on top of Java ExecutorService framework.

1. Spring @EnableAsync and @Async

Spring comes with @EnableAsync annotation and can be applied to a @Configuration class for asynchronous behavior. The @EnableAsync annotation will look for methods marked with @Async annotation and run them in background thread pools.

The @Async annotated methods are executed in a separate thread and return CompletableFuture to hold the result of an asynchronous computation.

To enable async configuration in spring, follow these steps:

• Create a thread pool to run the tasks asynchronously.

@Configuration
@EnableAsync
public class AsyncConfiguration {

  @Bean(name = "asyncExecutor")
  public Executor asyncExecutor()  {

    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(3);
    executor.setMaxPoolSize(3);
    executor.setQueueCapacity(100);
    executor.setThreadNamePrefix("AsynchThread-");
    executor.initialize();
    return executor;
  }
}

• Annotate the method with @Async that shall run asynchronously. The method must be public and may or may not return a value. The return value should be wrapped in a Future interface implementation if it returns a value.

@Async("asyncExecutor")
public CompletableFuture<EmployeeNames> methodOne() throws InterruptedException {
  //code
}

• To combine the result of multiple async tasks, use join() method.

CompletableFuture.allOf(methodOne, methodTwo, methodThree).join();

2. Spring REST Controller Example with Async Tasks

In this demo, we will create a REST API that will fetch data from three remote services asynchronously and when responses from all 3 services are available, then aggregate the responses.

• Invoke EmployeeName API
• Invoke EmployeeAddress API
• Invoke EmployeePhone API
• Wait for responses from the above services
• Aggregate all three API responses and build the final response to send back to the client

2.1. Remote REST APIs to be Consumed Asynchronously

The followings are remote APIs that the async REST controller must consume before aggregating the data and returning the result,

@RestController
public class EmployeeDataController {

  @RequestMapping(value = "/addresses", method = RequestMethod.GET)
  public EmployeeAddresses getAddresses() {
    //...
  }
 
  @RequestMapping(value = "/phones", method = RequestMethod.GET)
  public EmployeePhone getPhoneNumbers() {
    //...
  }
 
  @RequestMapping(value = "/names", method = RequestMethod.GET)
  public EmployeeNames getEmployeeName() {
    //...
  }
}

2.3. @Async Methods Returning CompletableFuture

These service methods will pull the data from the remote APIs or a database and must run in parallel in separate threads to speed up the process.

@Service
public class AsyncService {
 
  private static Logger log = LoggerFactory.getLogger(AsyncService.class);
 
  @Autowired
  private RestTemplate restTemplate;
 
  @Bean
  public RestTemplate restTemplate() {
    return new RestTemplate();
  }
 
  @Async
  public CompletableFuture<EmployeeNames> getEmployeeName() throws InterruptedException 
  {
    log.info("getEmployeeName starts");
 
    EmployeeNames employeeNameData = restTemplate.getForObject("http://localhost:8080/name", EmployeeNames.class);
 
    log.info("employeeNameData, {}", employeeNameData);
    Thread.sleep(1000L);  //Intentional delay
    log.info("employeeNameData completed");
    return CompletableFuture.completedFuture(employeeNameData);
  }
 
  @Async
  public CompletableFuture<EmployeeAddresses> getEmployeeAddress() throws InterruptedException 
  {
    log.info("getEmployeeAddress starts");
 
    EmployeeAddresses employeeAddressData = restTemplate.getForObject("http://localhost:8080/address", EmployeeAddresses.class);
 
    log.info("employeeAddressData, {}", employeeAddressData);
    Thread.sleep(1000L);  //Intentional delay
    log.info("employeeAddressData completed");
    return CompletableFuture.completedFuture(employeeAddressData);
  }
 
  @Async
  public CompletableFuture<EmployeePhone> getEmployeePhone() throws InterruptedException 
  {
    log.info("getEmployeePhone starts");
 
    EmployeePhone employeePhoneData = restTemplate.getForObject("http://localhost:8080/phone", EmployeePhone.class);
 
    log.info("employeePhoneData, {}", employeePhoneData);
    Thread.sleep(1000L);  //Intentional delay
    log.info("employeePhoneData completed");
    return CompletableFuture.completedFuture(employeePhoneData);
  }
}

2.3. Invoking Async Methods and Aggregating Results

This is the main API that calls the async methods, consumes and aggregates their responses, and returns to the client.

@RestController
public class AsyncController {
 
  private static Logger log = LoggerFactory.getLogger(AsyncController.class);
 
  @Autowired
  private AsyncService service;
 
  @RequestMapping(value = "/testAsynch", method = RequestMethod.GET)
  public void testAsynch() throws InterruptedException, ExecutionException 
  {
    log.info("testAsynch Start");
 
    CompletableFuture<EmployeeAddresses> employeeAddress = service.getEmployeeAddress();
    CompletableFuture<EmployeeNames> employeeName = service.getEmployeeName();
    CompletableFuture<EmployeePhone> employeePhone = service.getEmployeePhone();
 
    // Wait until they are all done
    CompletableFuture.allOf(employeeAddress, employeeName, employeePhone).join();
     
    log.info("EmployeeAddress--> " + employeeAddress.get());
    log.info("EmployeeName--> " + employeeName.get());
    log.info("EmployeePhone--> " + employeePhone.get());
  }
}

3. Exception Handling

When a method return type is a Future, the Future.get() method will throw the exception and we should use try-catch block to catch and handle the exception before aggregating the results.

The problem is if the async method does not return any value then it is hard to know if an exception occurred while the method was executing. We can use AsyncUncaughtExceptionHandler implementation for catching and handling such exceptions.

@Configuration
@EnableAsync
public class AsyncConfig extends AsyncConfigurerSupport {

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return new AsyncExceptionHandler();
    }
}

public class AsyncExceptionHandler implements AsyncUncaughtExceptionHandler {

  private final Logger logger = LoggerFactory.getLogger(AsyncExceptionHandler.class);

  @Override
  public void handleUncaughtException(Throwable ex, Method method, Object... params) {
      logger.error("Unexpected asynchronous exception at : "
              + method.getDeclaringClass().getName() + "." + method.getName(), ex);
  }
}

4. Run the Demo

Download and start both applications. Hit the API: http://localhost:8081/testAsynch. Observe the output in the console.

With @Async

Without @Async

Drop me your questions related to creating a spring boot non-blocking rest api.

Happy Learning !!

Source Code on Github