Spring Boot @Async — The Complete Guide

@SpringBootApplication
@EnableAsync
public class ECommerceApplication {
    public static void main(String[] args) {
        SpringApplication.run(ECommerceApplication.class, args);
    }
}

// Force CGLIB proxies (class-based, works even without interfaces)
@EnableAsync(proxyTargetClass = true)

// Use AspectJ weaving instead of proxies (compile-time, no self-invocation problem)
@EnableAsync(mode = AdviceMode.ASPECTJ)

@Configuration
@EnableAsync
public class AsyncConfig implements AsyncConfigurer {

    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(8);          // Always-alive threads
        executor.setMaxPoolSize(20);          // Burst capacity
        executor.setQueueCapacity(200);       // Bounded! OOM protection
        executor.setThreadNamePrefix("async-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setAwaitTerminationSeconds(60);
        executor.initialize();
        return executor;
    }

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

@Configuration
@EnableAsync
public class MultiExecutorConfig {

    @Bean("emailExecutor")
    public Executor emailExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(2);
        executor.setMaxPoolSize(5);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("email-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setAwaitTerminationSeconds(30);
        executor.initialize();
        return executor;
    }

    @Bean("reportExecutor")
    public Executor reportExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(4);
        executor.setMaxPoolSize(8);
        executor.setQueueCapacity(50);
        executor.setThreadNamePrefix("report-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setAwaitTerminationSeconds(120);  // Reports take time
        executor.initialize();
        return executor;
    }

    @Bean("webhookExecutor")
    public Executor webhookExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(30);
        executor.setQueueCapacity(500);
        executor.setThreadNamePrefix("webhook-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }
}

// Usage — specify which executor
@Async("emailExecutor")
public void sendOrderConfirmation(Order order) { ... }

@Async("reportExecutor")
public CompletableFuture<byte[]> generateMonthlyReport(Long accountId) { ... }

@Async("webhookExecutor")
public void processPaymentWebhook(WebhookEvent event) { ... }

@Async("emailExecutor")
public void sendOrderConfirmationEmail(Order order) {
    log.info("[{}] Sending confirmation for order {}",
        Thread.currentThread().getName(), order.getId());

    emailClient.send(
        order.getCustomerEmail(),
        "Order #" + order.getId() + " Confirmed",
        templateEngine.render("order-confirmation", Map.of("order", order))
    );
}

@Async
public Future<InventoryReport> generateInventoryReport(Long warehouseId) {
    InventoryReport report = inventoryService.calculateStock(warehouseId);
    return new AsyncResult<>(report);  // Spring's Future wrapper
}

// Caller — blocks on get()
Future<InventoryReport> future = reportService.generateInventoryReport(42L);
InventoryReport report = future.get(30, TimeUnit.SECONDS);  // Blocks up to 30s

@Async("reportExecutor")
public CompletableFuture<SalesReport> generateSalesReport(Long storeId, DateRange range) {
    log.info("Generating sales report for store {} on thread {}",
        storeId, Thread.currentThread().getName());

    SalesReport report = analyticsEngine.computeSales(storeId, range);
    return CompletableFuture.completedFuture(report);
}

@Service
@RequiredArgsConstructor
public class DashboardService {

    private final UserService userService;
    private final OrderService orderService;
    private final RecommendationService recommendationService;

    public DashboardResponse buildDashboard(Long userId) {
        // All three run in parallel on separate threads
        CompletableFuture<UserProfile> profileFuture =
            userService.fetchProfile(userId);
        CompletableFuture<List<Order>> ordersFuture =
            orderService.fetchRecentOrders(userId);
        CompletableFuture<List<Product>> recommendationsFuture =
            recommendationService.getRecommendations(userId);

        // Wait for all to complete (with timeout)
        CompletableFuture.allOf(profileFuture, ordersFuture, recommendationsFuture)
            .orTimeout(5, TimeUnit.SECONDS)
            .join();

        return DashboardResponse.builder()
            .profile(profileFuture.join())
            .recentOrders(ordersFuture.join())
            .recommendations(recommendationsFuture.join())
            .build();
    }
}

@Service
public class OrderService {

    public void placeOrder(Order order) {
        orderRepository.save(order);
        sendConfirmation(order);  // ❌ NOT async! Calls this.sendConfirmation()
    }

    @Async
    public void sendConfirmation(Order order) {
        // This runs on the SAME thread as placeOrder()
        emailClient.send(order.getEmail(), "Your order is confirmed");
    }
}

@Async
public void processPaymentWebhook(WebhookEvent event) {
    // If this throws, the exception goes to... nowhere useful
    paymentService.reconcile(event.getPaymentId());
}

@Slf4j
public class CustomAsyncExceptionHandler implements AsyncUncaughtExceptionHandler {

    private final MeterRegistry meterRegistry;
    private final AlertService alertService;

    public CustomAsyncExceptionHandler(MeterRegistry meterRegistry, AlertService alertService) {
        this.meterRegistry = meterRegistry;
        this.alertService = alertService;
    }

    @Override
    public void handleUncaughtException(Throwable ex, Method method, Object... params) {
        String className = method.getDeclaringClass().getSimpleName();
        String methodName = method.getName();

        // 1. Log with full context
        log.error("Async exception in {}.{}() with params {}: {}",
            className, methodName, Arrays.toString(params), ex.getMessage(), ex);

        // 2. Increment failure metric
        meterRegistry.counter("async.exceptions",
            "class", className,
            "method", methodName,
            "exception", ex.getClass().getSimpleName()
        ).increment();

        // 3. Alert if critical
        if (ex instanceof PaymentException || ex instanceof DataIntegrityException) {
            alertService.sendPagerDuty(
                "Critical async failure in " + className + "." + methodName, ex);
        }
    }
}

@Async("webhookExecutor")
public CompletableFuture<PaymentResult> processPayment(PaymentRequest request) {
    try {
        PaymentResult result = paymentGateway.charge(request);
        return CompletableFuture.completedFuture(result);
    } catch (PaymentDeclinedException ex) {
        return CompletableFuture.failedFuture(ex);  // Java 9+
    }
}

// Caller-side error handling with recovery
paymentService.processPayment(request)
    .thenApply(result -> {
        orderService.markPaid(orderId, result.getTransactionId());
        return result;
    })
    .exceptionally(ex -> {
        log.error("Payment failed for order {}: {}", orderId, ex.getMessage());
        orderService.markPaymentFailed(orderId);
        notificationService.alertCustomer(customerId, "Payment failed");
        return PaymentResult.failed(ex.getMessage());
    });

@Service
@RequiredArgsConstructor
public class OrderService {

    private final OrderRepository orderRepository;
    private final InventoryService inventoryService;

    @Transactional
    public void placeOrder(Order order) {
        orderRepository.save(order);          // Inside transaction
        inventoryService.syncAsync(order);    // New thread = NO transaction from parent
        // If syncAsync fails, this transaction is NOT rolled back
    }
}

@Service
public class InventoryService {

    @Async
    @Transactional  // Starts its OWN independent transaction
    public void syncAsync(Order order) {
        // This runs in a completely separate transaction
        // If it fails, the order is already committed
        inventoryRepository.decrementStock(order.getProductId(), order.getQuantity());
    }
}

@Service
@RequiredArgsConstructor
public class OrderService {

    private final ApplicationEventPublisher eventPublisher;

    @Transactional
    public Order placeOrder(OrderRequest request) {
        Order order = orderRepository.save(new Order(request));
        // Event published AFTER transaction commits
        eventPublisher.publishEvent(new OrderPlacedEvent(order.getId()));
        return order;
    }
}

@Component
@RequiredArgsConstructor
public class OrderEventHandler {

    @Async("emailExecutor")
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void onOrderPlaced(OrderPlacedEvent event) {
        // Guaranteed: order is committed before this runs
        // Runs on async thread, won't block the response
        Order order = orderRepository.findById(event.getOrderId()).orElseThrow();
        emailService.sendConfirmation(order);
        inventoryService.syncWithWarehouse(order);
    }
}

@Slf4j
public class MdcTaskDecorator implements TaskDecorator {

    @Override
    public Runnable decorate(Runnable runnable) {
        // Capture context on the submitting thread
        Map<String, String> mdcContext = MDC.getCopyOfContextMap();
        RequestAttributes requestAttributes =
            RequestContextHolder.getRequestAttributes();

        return () -> {
            try {
                // Set context on the executing thread
                if (mdcContext != null) {
                    MDC.setContextMap(mdcContext);
                }
                if (requestAttributes != null) {
                    RequestContextHolder.setRequestAttributes(requestAttributes);
                }
                runnable.run();
            } finally {
                // Clean up to prevent thread pool pollution
                MDC.clear();
                RequestContextHolder.resetRequestAttributes();
            }
        };
    }
}

@Bean("taskExecutor")
public Executor taskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(8);
    executor.setMaxPoolSize(20);
    executor.setQueueCapacity(200);
    executor.setThreadNamePrefix("async-");
    executor.setTaskDecorator(new MdcTaskDecorator());  // Propagate MDC
    executor.initialize();
    return executor;
}

@Bean("secureExecutor")
public Executor secureExecutor() {
    ThreadPoolTaskExecutor delegate = new ThreadPoolTaskExecutor();
    delegate.setCorePoolSize(5);
    delegate.setMaxPoolSize(10);
    delegate.setQueueCapacity(100);
    delegate.setThreadNamePrefix("secure-async-");
    delegate.initialize();

    // Wraps every task to copy SecurityContext from submitter → executor
    return new DelegatingSecurityContextAsyncTaskExecutor(delegate);
}

@Async("secureExecutor")
public void auditUserAction(String action) {
    // SecurityContextHolder.getContext().getAuthentication() works here
    String username = SecurityContextHolder.getContext()
        .getAuthentication().getName();
    auditRepository.save(new AuditEntry(username, action, Instant.now()));
}

@Service
@Slf4j
public class PaymentNotificationService {

    @Async("webhookExecutor")
    @Retryable(
        retryFor = {RestClientException.class, TimeoutException.class},
        maxAttempts = 3,
        backoff = @Backoff(delay = 2000, multiplier = 2)  // 2s, 4s, 8s
    )
    public CompletableFuture<Void> notifyMerchant(PaymentEvent event) {
        log.info("Notifying merchant {} for payment {} (thread: {})",
            event.getMerchantId(), event.getPaymentId(),
            Thread.currentThread().getName());

        restClient.post()
            .uri(event.getWebhookUrl())
            .body(event)
            .retrieve()
            .toBodilessEntity();

        return CompletableFuture.completedFuture(null);
    }

    @Recover
    public CompletableFuture<Void> notifyMerchantFallback(
            RestClientException ex, PaymentEvent event) {
        log.error("All retries exhausted for merchant {}, payment {}",
            event.getMerchantId(), event.getPaymentId(), ex);
        deadLetterQueue.push(event);  // Save for manual retry
        alertService.notify("Merchant webhook failed: " + event.getMerchantId());
        return CompletableFuture.completedFuture(null);
    }
}

@Service
@RequiredArgsConstructor
@Slf4j
public class BulkEmailService {

    private final EmailSender emailSender;

    @Async("emailExecutor")
    public CompletableFuture<BatchResult> sendBatch(List<EmailRequest> batch, int batchNumber) {
        log.info("Processing batch {} ({} emails) on thread {}",
            batchNumber, batch.size(), Thread.currentThread().getName());

        int success = 0, failed = 0;
        for (EmailRequest email : batch) {
            try {
                emailSender.send(email);
                success++;
            } catch (Exception ex) {
                failed++;
                log.warn("Failed to send email to {}: {}", email.getTo(), ex.getMessage());
            }
        }

        return CompletableFuture.completedFuture(new BatchResult(batchNumber, success, failed));
    }

    public CompletableFuture<List<BatchResult>> sendBulk(List<EmailRequest> allEmails) {
        // Partition into batches of 100
        List<List<EmailRequest>> batches = Lists.partition(allEmails, 100);

        List<CompletableFuture<BatchResult>> futures = IntStream.range(0, batches.size())
            .mapToObj(i -> sendBatch(batches.get(i), i))
            .toList();

        return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
            .thenApply(v -> futures.stream()
                .map(CompletableFuture::join)
                .toList());
    }
}

@Async("reportExecutor")
public CompletableFuture<Report> generateReport(ReportRequest request) {
    return CompletableFuture.supplyAsync(() -> {
        return heavyReportGeneration(request);
    }).orTimeout(30, TimeUnit.SECONDS);  // Java 9+
}

// Caller handles timeout gracefully
reportService.generateReport(request)
    .thenAccept(report -> cache.put(request.getId(), report))
    .exceptionally(ex -> {
        if (ex.getCause() instanceof TimeoutException) {
            log.warn("Report generation timed out for request {}", request.getId());
            notifyUser(request.getUserId(), "Report is taking longer than expected");
        }
        return null;
    });

@Bean("taskExecutor")
public Executor taskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(8);
    executor.setMaxPoolSize(20);
    executor.setQueueCapacity(200);
    executor.setThreadNamePrefix("async-");

    // Graceful shutdown: finish in-progress tasks before app dies
    executor.setWaitForTasksToCompleteOnShutdown(true);
    executor.setAwaitTerminationSeconds(60);  // Wait up to 60s

    executor.initialize();
    return executor;
}

@Bean("taskExecutor")
public Executor taskExecutor(MeterRegistry meterRegistry) {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(8);
    executor.setMaxPoolSize(20);
    executor.setQueueCapacity(200);
    executor.setThreadNamePrefix("async-");
    executor.initialize();

    // Expose metrics
    ExecutorServiceMetrics.monitor(
        meterRegistry,
        executor.getThreadPoolExecutor(),
        "async.task.executor"
    );

    return executor;
}

@Bean("virtualExecutor")
public Executor virtualExecutor() {
    return Executors.newVirtualThreadPerTaskExecutor();
}

@Async("virtualExecutor")
public CompletableFuture<OrderStatus> checkOrderStatus(Long orderId) {
    // Each call gets its own virtual thread — no pool exhaustion
    OrderStatus status = externalOrderApi.getStatus(orderId);
    return CompletableFuture.completedFuture(status);
}

@SpringBootTest
class OrderServiceIntegrationTest {

    @Autowired
    private OrderService orderService;

    @Autowired
    private EmailRepository emailRepository;

    @Test
    void placeOrder_sendsConfirmationEmailAsynchronously() {
        Order order = new Order("user@example.com", "Product-123");

        orderService.placeOrder(order);

        // Don't use Thread.sleep()! Use Awaitility
        await()
            .atMost(Duration.ofSeconds(5))
            .pollInterval(Duration.ofMillis(200))
            .until(() -> emailRepository.findByRecipient("user@example.com").isPresent());

        Email sent = emailRepository.findByRecipient("user@example.com").get();
        assertThat(sent.getSubject()).contains("Order Confirmed");
    }
}

@Configuration
@Profile("test")
public class TestAsyncConfig {

    @Bean("emailExecutor")
    @Primary
    public Executor syncExecutor() {
        // Makes @Async methods run synchronously in tests — deterministic, no timing issues
        return new SyncTaskExecutor();
    }
}

@Test
void asyncMethod_runsOnDifferentThread() {
    String callerThread = Thread.currentThread().getName();

    CompletableFuture<String> future = asyncService.doWork();
    String asyncThread = future.join();

    assertThat(asyncThread).isNotEqualTo(callerThread);
    assertThat(asyncThread).startsWith("async-");  // Matches threadNamePrefix
}

@Configuration
@EnableAsync
@Slf4j
public class AsyncConfiguration implements AsyncConfigurer {

    @Override
    public Executor getAsyncExecutor() {
        return buildExecutor("default-async-", 8, 20, 200);
    }

    @Bean("emailExecutor")
    public Executor emailExecutor() {
        return buildExecutor("email-", 2, 5, 100);
    }

    @Bean("inventoryExecutor")
    public Executor inventoryExecutor() {
        return buildExecutor("inventory-", 4, 10, 50);
    }

    @Bean("analyticsExecutor")
    public Executor analyticsExecutor() {
        return buildExecutor("analytics-", 2, 4, 500);  // Large queue, best-effort
    }

    private Executor buildExecutor(String prefix, int core, int max, int queue) {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(core);
        executor.setMaxPoolSize(max);
        executor.setQueueCapacity(queue);
        executor.setThreadNamePrefix(prefix);
        executor.setTaskDecorator(new MdcTaskDecorator());
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setAwaitTerminationSeconds(60);
        executor.initialize();
        return executor;
    }

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return (ex, method, params) -> {
            log.error("Uncaught async exception in {}.{}(): {}",
                method.getDeclaringClass().getSimpleName(),
                method.getName(), ex.getMessage(), ex);
        };
    }
}

@Service
@RequiredArgsConstructor
@Slf4j
public class OrderOrchestrator {

    private final OrderRepository orderRepository;
    private final ApplicationEventPublisher eventPublisher;

    @Transactional
    public OrderResponse placeOrder(OrderRequest request) {
        // 1. Save order (transactional)
        Order order = orderRepository.save(Order.from(request));

        // 2. Publish event — listeners run AFTER commit, asynchronously
        eventPublisher.publishEvent(new OrderPlacedEvent(order));

        // 3. Return immediately
        return OrderResponse.success(order.getId());
    }
}

@Component
@RequiredArgsConstructor
@Slf4j
public class OrderEventHandlers {

    private final EmailService emailService;
    private final InventoryService inventoryService;
    private final AnalyticsService analyticsService;

    @Async("emailExecutor")
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void sendConfirmation(OrderPlacedEvent event) {
        log.info("Sending order confirmation for {} on thread {}",
            event.getOrderId(), Thread.currentThread().getName());
        emailService.sendOrderConfirmation(event.getOrderId());
    }

    @Async("inventoryExecutor")
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void syncInventory(OrderPlacedEvent event) {
        log.info("Syncing inventory for order {} on thread {}",
            event.getOrderId(), Thread.currentThread().getName());
        inventoryService.decrementAndSync(event.getOrderId());
    }

    @Async("analyticsExecutor")
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void trackAnalytics(OrderPlacedEvent event) {
        analyticsService.trackPurchase(event.getOrderId());
    }
}

// 1. Enable
@EnableAsync

// 2. Configure (NEVER skip this)
@Bean public Executor taskExecutor() { /* ThreadPoolTaskExecutor */ }

// 3. Use
@Async                              // Default executor
@Async("emailExecutor")             // Named executor

// 4. Return types
@Async public void fire() {}                           // Fire & forget
@Async public CompletableFuture<T> compute() {}        // Composable result

// 5. Handle errors
implements AsyncConfigurer → getAsyncUncaughtExceptionHandler()
.exceptionally(ex -> fallback)

// 6. Propagate context
executor.setTaskDecorator(new MdcTaskDecorator())
new DelegatingSecurityContextAsyncTaskExecutor(executor)

// 7. Shutdown gracefully
executor.setWaitForTasksToCompleteOnShutdown(true)
executor.setAwaitTerminationSeconds(60)