Events, scheduling, @Async: làm việc ngoài luồng request

@Async là annotation của Spring (org.springframework.scheduling.annotation.Async). Bên dưới nó dùng thứ của Java core: thread pool, Executor, CompletableFuture.

1. Bài toán: tạo đơn xong phải gửi email, mất 5 giây

public Order createOrder(CreateOrderRequest req) {
    System.out.println("Create order: " + Thread.currentThread().getName());
    Order order = orderRepository.save(Order.from(req));   // 1s
    emailService.sendOrderConfirmation(order);              // 5s
    System.out.println("DONE");
    return order;
}

http-nio-8080-exec-1
 |
 | createOrder()
 | save order ---- 1s ---->
 | sendEmail() --------- 5s --------->
 | DONE
 v
User chờ ~6s

2. Vấn đề

• User chờ cả email, trong khi business chỉ cần "order tạo thành công là đủ, email gửi sau cũng được".
• Tomcat thread bị giữ 5 giây vô ích; nhiều request cùng lúc thì hết thread.
• SMTP lỗi làm cả request tạo đơn lỗi theo.

3. Giải pháp: @Async đẩy việc sang thread pool

Có thể hình dung Spring làm giúp bạn chuyện này:

executor.submit(() -> emailService.sendOrderConfirmation(order));

Java Core:   Thread → ExecutorService → Future → CompletableFuture
                                   ↓
Spring:                       TaskExecutor
                                   ↓
                                @Async

4. Code từng bước

Bước 1: Dependency. Có sẵn trong spring-context, không cần thêm gì.

Bước 2: Bật async và cấu hình executor riêng

@Configuration
@EnableAsync
public class AsyncConfig {

    @Bean(name = "emailExecutor")
    public ThreadPoolTaskExecutor emailExecutor() {
        var ex = new ThreadPoolTaskExecutor();
        ex.setCorePoolSize(4);
        ex.setMaxPoolSize(8);
        ex.setQueueCapacity(500);                 // đừng để vô hạn
        ex.setThreadNamePrefix("email-");
        ex.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        ex.setWaitForTasksToCompleteOnShutdown(true);
        ex.setAwaitTerminationSeconds(30);
        ex.initialize();
        return ex;
    }
}

Spring Boot cũng tự tạo một executor mặc định (applicationTaskExecutor), chỉnh qua spring.task.execution.pool.*. Đặt tên executor riêng giúp tách pool email khỏi pool việc khác.

Bước 3: EmailService

@Service
public class EmailService {

    @Async("emailExecutor")
    public void sendOrderConfirmation(Order order) {
        System.out.println("Send email: " + Thread.currentThread().getName());
        // gọi SMTP, mất 5s
    }
}

OrderService không đổi gì, vẫn gọi emailService.sendOrderConfirmation(order).

Bước 4: Test

time curl -X POST localhost:8080/orders -H "Content-Type: application/json" -d '{"productId":1,"quantity":1}'
# real 0m1.1s

# log:
Create order: http-nio-8080-exec-1
DONE
Send email: email-1

5. Cách hoạt động

Tomcat thread                         emailExecutor
http-nio-8080-exec-1
 |
 | createOrder()
 | emailService.sendOrderConfirmation()
 |    ↓ (gọi vào PROXY của EmailService)
 |    AsyncExecutionInterceptor
 |    +------ submit task ------------>  email-1
 | DONE                                    |
 | return 200 (~1s)                        | sendOrderConfirmation()
 v                                         | block 5s
                                           v
                                         thread quay về pool

Hai điều kiện để @Async chạy: có @EnableAsync, và lời gọi phải đi qua proxy. Gọi this.sendEmail() trong cùng class thì chạy đồng bộ (bài 4.1.3).

@Async không có nghĩa là non-blocking. JDBC, SMTP vẫn block. Nó chỉ chuyển "Tomcat thread bị block" thành "worker thread bị block".

6. So sánh kiểu trả về

7. Khi nào nên dùng / không nên dùng

• Nên: việc phụ, user không cần kết quả ngay: email, push notification, ghi log phân tích.
• Không nên: việc quan trọng không được mất. Task nằm trong RAM, server crash là mất. Dùng message queue (RabbitMQ, Kafka) hoặc outbox.
• Không gọi @Async rồi join() ngay trong request: user vẫn chờ, mà dùng tới 2 thread.

Caller không biết gì cả. Với @Async void thì caller submit xong là xong, không quan tâm nữa. Worker gửi xong thì thread quay về pool. Nhưng có một hệ quả quan trọng: nếu sendEmail lỗi thì sao?

Tomcat thread                 email-1
sendOrderConfirmation()
 +------------------------>   sending...
DONE
HTTP 200 ✔
                              💥 SMTP ERROR

HTTP 200 đã trả rồi. Exception của worker không thể quay ngược thời gian đổi thành 500. Với method void, exception đi vào AsyncUncaughtExceptionHandler; mặc định chỉ log. Nên cấu hình để ít nhất có log rõ ràng hoặc lưu lại để retry:

@Configuration
@EnableAsync
public class AsyncConfig implements AsyncConfigurer {

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return (ex, method, params) ->
            log.error("Async {} failed, params={}", method.getName(), params, ex);
    }
}

Đây chính là lúc CompletableFuture phát huy tác dụng: @Async đưa việc sang thread khác, CompletableFuture trả lời câu hỏi "xong rồi thì làm gì tiếp".

@Async("emailExecutor")
public CompletableFuture<Boolean> sendOrderConfirmation(Order order) {
    boolean success = smtp.send(order);
    return CompletableFuture.completedFuture(success);
}

Caller không join(), mà đăng ký continuation:

public Order createOrder(CreateOrderRequest req) {
    Order order = orderRepository.save(Order.from(req));
    emailService.sendOrderConfirmation(order)
        .thenAccept(ok -> emailHistoryService.save(order.getId(), ok))
        .exceptionally(ex -> { log.error("Email failed", ex); return null; });
    return order;                                  // user xong ở đây
}

User flow              Background flow
Create order
   |                   Send email (5s)
200 OK  ← user xong        |
                       completed
                           |
                       save email history

Ba case để nhớ:

Insight: async không có nghĩa là user không chờ. User có chờ hay không phụ thuộc request có cần kết quả trước khi trả response hay không. Lưu ý: thenAccept chạy trên thread worker, nên đừng dùng request scope hay SecurityContext trong đó mà không truyền sang.

Có: Spring Application Events. OrderService chỉ thông báo "đơn đã tạo", ai quan tâm thì nghe.

public record OrderCreatedEvent(Long orderId, String email) { }

@Service
public class OrderService {
    private final ApplicationEventPublisher publisher;
    ...
    @Transactional
    public Order createOrder(CreateOrderRequest req) {
        Order order = orderRepository.save(Order.from(req));
        publisher.publishEvent(new OrderCreatedEvent(order.getId(), req.email()));
        return order;
    }
}

@Component
public class OrderNotificationListener {

    @Async("emailExecutor")
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void onOrderCreated(OrderCreatedEvent e) {
        emailService.sendOrderConfirmation(e.orderId(), e.email());
    }
}

@Component
public class LoyaltyListener {
    @EventListener
    public void addPoints(OrderCreatedEvent e) { ... }   // chạy đồng bộ, trong transaction
}

createOrder() ── BEGIN TX
   save order
   publishEvent ──→ @EventListener LoyaltyListener (ngay, cùng thread, cùng TX)
                 ──→ @TransactionalEventListener: giữ lại, chờ...
COMMIT ✔ ─────────→ AFTER_COMMIT: OrderNotificationListener (async)
ROLLBACK ✘ ───────→ không gửi email

Vì sao email phải AFTER_COMMIT? Nếu gửi ngay mà transaction rollback, khách nhận email cho đơn không tồn tại. Lưu ý: event vẫn nằm trong RAM; muốn chắc chắn không mất thì dùng outbox pattern + message broker.

@Scheduled làm việc này rất gọn:

@Configuration
@EnableScheduling
public class SchedulingConfig { }

@Component
public class OrderCleanupJob {

    @Scheduled(cron = "0 */5 * * * *")      // mỗi 5 phút
    public void cancelUnpaidOrders() {
        int n = orderRepository.cancelUnpaidBefore(Instant.now().minus(Duration.ofMinutes(30)));
        log.info("Cancelled {} unpaid orders", n);
    }
}

Vấn đề: @Scheduled chạy trong mỗi instance. 3 pod = job chạy 3 lần cùng lúc, có thể gửi 3 email hủy đơn, hoặc tranh nhau lock DB.

pod-1 ──10:05──→ cancelUnpaidOrders()
pod-2 ──10:05──→ cancelUnpaidOrders()   ← trùng
pod-3 ──10:05──→ cancelUnpaidOrders()   ← trùng

Giải pháp phổ biến: ShedLock, dùng một bảng DB (hoặc Redis) làm distributed lock.

<dependency>
    <groupId>net.javacrumbs.shedlock</groupId>
    <artifactId>shedlock-spring</artifactId>
</dependency>
<dependency>
    <groupId>net.javacrumbs.shedlock</groupId>
    <artifactId>shedlock-provider-jdbc-template</artifactId>
</dependency>

@Configuration
@EnableScheduling
@EnableSchedulerLock(defaultLockAtMostFor = "10m")
public class SchedulingConfig {
    @Bean
    public LockProvider lockProvider(DataSource dataSource) {
        return new JdbcTemplateLockProvider(
            JdbcTemplateLockProvider.Configuration.builder()
                .withJdbcTemplate(new JdbcTemplate(dataSource))
                .usingDbTime()
                .build());
    }
}

@Scheduled(cron = "0 */5 * * * *")
@SchedulerLock(name = "cancelUnpaidOrders", lockAtMostFor = "4m", lockAtLeastFor = "30s")
public void cancelUnpaidOrders() { ... }

CREATE TABLE shedlock (
  name       VARCHAR(64)  PRIMARY KEY,
  lock_until TIMESTAMP    NOT NULL,
  locked_at  TIMESTAMP    NOT NULL,
  locked_by  VARCHAR(255) NOT NULL
);

10:05  pod-1 UPDATE shedlock SET lock_until=10:09 WHERE name=... AND lock_until <= now → 1 row ✔ chạy
       pod-2 UPDATE ... → 0 row → bỏ qua
       pod-3 UPDATE ... → 0 row → bỏ qua

• lockAtMostFor: pod chết giữa chừng thì lock tự hết hạn, không kẹt mãi.
• lockAtLeastFor: tránh pod khác chạy lại ngay nếu job xong quá nhanh và đồng hồ lệch.
• Job vẫn nên idempotent (chỉ hủy đơn còn UNPAID), vì lock không phải bảo đảm tuyệt đối.

Lưu ý thêm: mặc định scheduler chỉ có 1 thread, job chậm sẽ chặn job khác. Tăng spring.task.scheduling.pool.size. Job lớn, cần retry/lịch sử thì cân nhắc Quartz cluster hoặc Kubernetes CronJob.

Q1. @Async hoạt động thế nào? Có bẫy gì?

🇻🇳 Tiếng Việt:

@Async là annotation của Spring, cần @EnableAsync. Spring bọc bean bằng proxy; khi gọi method, proxy submit việc sang TaskExecutor và trả về ngay. Bẫy: gọi trong cùng class không qua proxy nên chạy đồng bộ; method void mất exception nếu không cấu hình handler; nên dùng executor riêng có giới hạn queue; và nó không biến I/O blocking thành non-blocking.

🇬🇧 English:

@Async is a Spring annotation enabled by @EnableAsync. Spring wraps the bean in a proxy; on each call the proxy submits the work to a TaskExecutor and returns immediately. Pitfalls: self-invocation bypasses the proxy and runs synchronously, exceptions from void methods are lost unless you configure a handler, you should use a dedicated bounded executor, and it does not make blocking I/O non-blocking.

Q2. @EventListener và @TransactionalEventListener khác nhau thế nào?

🇻🇳 Tiếng Việt:

@EventListener chạy ngay khi publish, đồng bộ, cùng transaction. @TransactionalEventListener chờ transaction tới phase chỉ định, mặc định là sau commit. Gửi email hay gọi hệ thống ngoài nên dùng AFTER_COMMIT để không gửi thông báo cho dữ liệu bị rollback.

🇬🇧 English:

@EventListener runs synchronously as soon as the event is published, inside the same transaction. @TransactionalEventListener waits for a transaction phase, AFTER_COMMIT by default. External side effects like emails should run after commit so you never notify about data that was rolled back.

Q3. @Scheduled chạy trên nhiều instance thì xử lý sao?

🇻🇳 Tiếng Việt:

Mỗi instance đều chạy job nên bị trùng. Dùng distributed lock như ShedLock với bảng DB hoặc Redis để chỉ một instance chạy mỗi lần, đặt lockAtMostFor để lock tự hết hạn khi pod chết, và làm job idempotent. Job phức tạp thì dùng Quartz cluster hoặc Kubernetes CronJob.

🇬🇧 English:

Every instance runs the job, so it executes multiple times. Use a distributed lock such as ShedLock backed by a database table or Redis so only one instance runs it, set lockAtMostFor so the lock expires if a pod dies, and keep the job idempotent. For complex jobs, use Quartz in cluster mode or a Kubernetes CronJob.

Cách nhớ ngắn gọn

• @Async = proxy + thread pool. Cần @EnableAsync, cấm self-call.
• void = fire-and-forget; CompletableFuture = "xong rồi làm gì tiếp".
• Side effect ra ngoài → @TransactionalEventListener(AFTER_COMMIT).
• N instance × @Scheduled = N lần chạy → ShedLock.
• Không được mất việc → queue/outbox, không phải @Async.