Spring Boot Async Processing & Scheduled Tasks

Giới Thiệu

Trong các ứng dụng thực tế, không phải mọi thao tác đều cần xảy ra đồng bộ. Gửi email, tạo báo cáo, xử lý tải file, và đồng bộ dữ liệu với hệ thống bên ngoài là những tác vụ có thể — và nên — chạy trong nền.

Spring Boot cung cấp hỗ trợ mạnh mẽ tích hợp sẵn cho xử lý bất đồng bộ thông qua annotation @Async và CompletableFuture, cùng với thực thi tác vụ theo lịch với @Scheduled. Kết hợp lại, những công cụ này cho phép bạn xây dựng các ứng dụng phản hồi nhanh, hiệu quả, xử lý các thao tác kéo dài mà không chặn request của người dùng.

Những Gì Bạn Sẽ Học:

✅ Bật và cấu hình xử lý async với @Async
✅ Trả về kết quả từ phương thức async với CompletableFuture
✅ Cấu hình thread pool tùy chỉnh cho production
✅ Scheduled tasks với @Scheduled và cron expressions
✅ Xử lý lỗi trong quy trình async
✅ Kết hợp nhiều thao tác async
✅ Xây dựng dịch vụ thông báo nền

Yêu Cầu:

Đã hoàn thành Bắt Đầu với Spring Boot
Hiểu biết cơ bản về REST APIs và Spring Services
Đã cài đặt Java 17+

Tại Sao Cần Xử Lý Async?

Hãy xem xét một quy trình đăng ký người dùng điển hình:

Đồng bộ (chậm):
  Request → Lưu User → Gửi Email → Gửi SMS → Ghi Log → Response
  Tổng: 50ms + 500ms + 300ms + 100ms = 950ms
 
Bất đồng bộ (nhanh):
  Request → Lưu User → Response (50ms)
             ├── Gửi Email (nền)
             ├── Gửi SMS (nền)
             └── Ghi Log (nền)
  Tổng: ~50ms (người dùng nhận phản hồi ngay lập tức)

Xử lý async cải thiện:

Thời gian phản hồi — Người dùng nhận phản hồi ngay lập tức
Throughput — Các thread server không bị chặn khi chờ I/O chậm
Khả năng chịu lỗi — Tác vụ nền có thể retry độc lập khi thất bại
Khả năng mở rộng — Thread pool riêng biệt cô lập các workload

Thiết Lập Dự Án

Bước 1: Thêm Dependencies

Chúng ta sẽ sử dụng một dự án Spring Boot web tiêu chuẩn. Thêm vào pom.xml:

<dependencies>
    <!-- Spring Boot Starter Web -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
 
    <!-- Spring Boot Starter Mail (cho ví dụ email) -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-mail</artifactId>
    </dependency>
 
    <!-- Lombok -->
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <optional>true</optional>
    </dependency>
 
    <!-- Spring Boot Starter Test -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

Bước 2: Cấu Hình Application Properties

Cập nhật application.yml:

spring:
  application:
    name: async-demo
 
  # Cấu hình Mail (cho ví dụ notification)
  mail:
    host: smtp.gmail.com
    port: 587
    username: ${MAIL_USERNAME:[email protected]}
    password: ${MAIL_PASSWORD:password}
    properties:
      mail.smtp.auth: true
      mail.smtp.starttls.enable: true
 
logging:
  level:
    com.example.async: DEBUG
    org.springframework.scheduling: DEBUG

Bật Xử Lý Async

Bước 1: Bật Hỗ Trợ @Async

Tạo một class cấu hình để bật xử lý async:

package com.example.async.config;
 
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
 
@Configuration
@EnableAsync
public class AsyncConfig {
    // @EnableAsync kích hoạt khả năng thực thi async của Spring
    // Mặc định, nó sử dụng SimpleAsyncTaskExecutor (không tái sử dụng thread)
    // Chúng ta sẽ cấu hình thread pool phù hợp ở bước tiếp theo
}

Bước 2: Tạo Async Service Cơ Bản

package com.example.async.service;
 
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
 
@Service
@Slf4j
public class NotificationService {
 
    @Async
    public void sendEmail(String to, String subject, String body) {
        log.info("Sending email to {} on thread: {}", to, Thread.currentThread().getName());
        try {
            // Mô phỏng gửi email chậm
            Thread.sleep(2000);
            log.info("Email sent successfully to {}", to);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("Email sending interrupted for {}", to);
        }
    }
 
    @Async
    public void sendSms(String phoneNumber, String message) {
        log.info("Sending SMS to {} on thread: {}", phoneNumber, Thread.currentThread().getName());
        try {
            Thread.sleep(1000);
            log.info("SMS sent successfully to {}", phoneNumber);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("SMS sending interrupted for {}", phoneNumber);
        }
    }
}

Bước 3: Gọi Phương Thức Async Từ Controller

package com.example.async.controller;
 
import com.example.async.service.NotificationService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
 
import java.util.Map;
 
@RestController
@RequestMapping("/api/users")
@RequiredArgsConstructor
@Slf4j
public class UserController {
 
    private final NotificationService notificationService;
 
    @PostMapping("/register")
    public ResponseEntity<Map<String, String>> registerUser(@RequestBody Map<String, String> user) {
        String email = user.get("email");
        String phone = user.get("phone");
        String name = user.get("name");
 
        log.info("Registering user: {} on thread: {}", name, Thread.currentThread().getName());
 
        // Lưu user vào database (đồng bộ)
        // userRepository.save(user);
 
        // Gửi thông báo (bất đồng bộ - trả về ngay lập tức)
        notificationService.sendEmail(email, "Welcome!", "Welcome to our platform, " + name);
        notificationService.sendSms(phone, "Your account has been created.");
 
        log.info("Registration complete for {} (notifications sent in background)", name);
 
        return ResponseEntity.ok(Map.of(
            "message", "User registered successfully",
            "email", email
        ));
    }
}

Kiểm tra:

curl -X POST http://localhost:8080/api/users/register \
  -H "Content-Type: application/json" \
  -d '{"name":"John","email":"[email protected]","phone":"+1234567890"}'

Log output mong đợi:

Registering user: John on thread: http-nio-8080-exec-1
Registration complete for John (notifications sent in background)
Sending email to [email protected] on thread: task-1
Sending SMS to +1234567890 on thread: task-2
Email sent successfully to [email protected]
SMS sent successfully to +1234567890

Lưu ý response trả về trước khi email và SMS được gửi. Các thông báo chạy trên các thread khác nhau (task-1, task-2).

Quan trọng: Phương thức @Async phải được gọi từ một class khác. Gọi phương thức @Async từ trong cùng class sẽ bỏ qua proxy và chạy đồng bộ. Điều này vì Spring sử dụng AOP proxy để chặn cuộc gọi và chuyển nó sang thread riêng.

Trả Về Kết Quả Với CompletableFuture

Phương thức @Async có thể trả về CompletableFuture<T> khi bạn cần kết quả sau:

Async Service Với Giá Trị Trả Về

package com.example.async.service;
 
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
 
import java.util.concurrent.CompletableFuture;
 
@Service
@Slf4j
public class DataAggregationService {
 
    @Async
    public CompletableFuture<String> fetchUserProfile(Long userId) {
        log.info("Fetching user profile for {} on thread: {}", userId, Thread.currentThread().getName());
        try {
            Thread.sleep(1000); // Mô phỏng gọi API bên ngoài
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return CompletableFuture.completedFuture("User-" + userId + " profile data");
    }
 
    @Async
    public CompletableFuture<String> fetchUserOrders(Long userId) {
        log.info("Fetching orders for {} on thread: {}", userId, Thread.currentThread().getName());
        try {
            Thread.sleep(1500); // Mô phỏng truy vấn database
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return CompletableFuture.completedFuture("User-" + userId + " has 5 orders");
    }
 
    @Async
    public CompletableFuture<String> fetchUserRecommendations(Long userId) {
        log.info("Fetching recommendations for {} on thread: {}", userId, Thread.currentThread().getName());
        try {
            Thread.sleep(2000); // Mô phỏng gọi ML service
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return CompletableFuture.completedFuture("3 recommended products for User-" + userId);
    }
}

Kết Hợp Nhiều Kết Quả Async

package com.example.async.controller;
 
import com.example.async.service.DataAggregationService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
 
import java.util.Map;
import java.util.concurrent.CompletableFuture;
 
@RestController
@RequestMapping("/api/dashboard")
@RequiredArgsConstructor
@Slf4j
public class DashboardController {
 
    private final DataAggregationService dataService;
 
    @GetMapping("/{userId}")
    public ResponseEntity<Map<String, String>> getUserDashboard(@PathVariable Long userId) throws Exception {
        long start = System.currentTimeMillis();
 
        // Chạy cả ba cuộc gọi song song
        CompletableFuture<String> profileFuture = dataService.fetchUserProfile(userId);
        CompletableFuture<String> ordersFuture = dataService.fetchUserOrders(userId);
        CompletableFuture<String> recommendationsFuture = dataService.fetchUserRecommendations(userId);
 
        // Chờ tất cả hoàn thành
        CompletableFuture.allOf(profileFuture, ordersFuture, recommendationsFuture).join();
 
        long duration = System.currentTimeMillis() - start;
        log.info("Dashboard loaded in {}ms (sequential would be ~4500ms)", duration);
 
        return ResponseEntity.ok(Map.of(
            "profile", profileFuture.get(),
            "orders", ordersFuture.get(),
            "recommendations", recommendationsFuture.get(),
            "loadTimeMs", String.valueOf(duration)
        ));
    }
}

So sánh hiệu suất:

Tuần tự:       1000ms + 1500ms + 2000ms = 4500ms
Song song async: max(1000, 1500, 2000)   = ~2000ms (nhanh hơn 2.25 lần)

Chuỗi Thao Tác CompletableFuture

@GetMapping("/{userId}/summary")
public CompletableFuture<ResponseEntity<Map<String, String>>> getUserSummary(@PathVariable Long userId) {
    return dataService.fetchUserProfile(userId)
        .thenCombine(dataService.fetchUserOrders(userId), (profile, orders) -> {
            // Kết hợp profile + orders
            return Map.of("profile", profile, "orders", orders);
        })
        .thenApply(result -> {
            // Chuyển đổi kết quả kết hợp
            return ResponseEntity.ok(result);
        })
        .exceptionally(ex -> {
            // Xử lý lỗi
            return ResponseEntity.internalServerError()
                .body(Map.of("error", ex.getMessage()));
        });
}

Cấu Hình Thread Pool Tùy Chỉnh

SimpleAsyncTaskExecutor mặc định tạo thread mới cho mỗi tác vụ — điều này không phù hợp cho production. Hãy cấu hình thread pool đúng cách:

Thread Pool Sẵn Sàng Production

package com.example.async.config;
 
import lombok.extern.slf4j.Slf4j;
import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.AsyncConfigurer;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
import java.util.concurrent.Executor;
 
@Configuration
@EnableAsync
@Slf4j
public class AsyncConfig implements AsyncConfigurer {
 
    @Override
    @Bean(name = "taskExecutor")
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
 
        // Core pool size: số thread duy trì ngay cả khi idle
        executor.setCorePoolSize(5);
 
        // Max pool size: số thread tối đa khi queue đầy
        executor.setMaxPoolSize(10);
 
        // Queue capacity: tác vụ chờ khi core threads bận
        executor.setQueueCapacity(100);
 
        // Tiền tố tên thread để debug dễ hơn
        executor.setThreadNamePrefix("async-");
 
        // Xử lý khi pool và queue đều đầy
        executor.setRejectedExecutionHandler((r, e) ->
            log.error("Task rejected: thread pool and queue are full")
        );
 
        // Chờ tác vụ hoàn thành khi shutdown
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setAwaitTerminationSeconds(30);
 
        executor.initialize();
        return executor;
    }
 
    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return (ex, method, params) ->
            log.error("Async error in method {}: {}", method.getName(), ex.getMessage(), ex);
    }
}

Nhiều Thread Pool Cho Các Workload Khác Nhau

Cô lập các workload khác nhau với pool riêng biệt:

@Configuration
@EnableAsync
public class AsyncConfig {
 
    @Bean(name = "emailExecutor")
    public Executor emailExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(2);
        executor.setMaxPoolSize(5);
        executor.setQueueCapacity(50);
        executor.setThreadNamePrefix("email-");
        executor.initialize();
        return executor;
    }
 
    @Bean(name = "reportExecutor")
    public Executor reportExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(3);
        executor.setMaxPoolSize(8);
        executor.setQueueCapacity(25);
        executor.setThreadNamePrefix("report-");
        executor.initialize();
        return executor;
    }
 
    @Bean(name = "notificationExecutor")
    public Executor notificationExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(2);
        executor.setMaxPoolSize(4);
        executor.setQueueCapacity(200);
        executor.setThreadNamePrefix("notify-");
        executor.initialize();
        return executor;
    }
}

Tham chiếu pool cụ thể với giá trị @Async:

@Async("emailExecutor")
public void sendEmail(String to, String subject, String body) {
    // Chạy trên emailExecutor pool
}
 
@Async("reportExecutor")
public CompletableFuture<byte[]> generateReport(Long reportId) {
    // Chạy trên reportExecutor pool
    return CompletableFuture.completedFuture(new byte[0]);
}

Hướng Dẫn Kích Thước Thread Pool

Loại Workload	Core Size	Max Size	Queue	Ví Dụ
CPU-bound	# CPUs	# CPUs	Nhỏ (10-50)	Xử lý dữ liệu
I/O-bound	# CPUs × 2	# CPUs × 4	Lớn (100-500)	HTTP calls, email
Hỗn hợp	# CPUs	# CPUs × 2	Trung bình (50-100)	Async chung

Scheduled Tasks Với @Scheduled

Annotation @Scheduled của Spring cho phép bạn chạy phương thức theo khoảng thời gian cố định hoặc lịch cron.

Bước 1: Bật Scheduling

package com.example.async.config;
 
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableScheduling;
 
@Configuration
@EnableScheduling
public class SchedulingConfig {
}

Bước 2: Tạo Scheduled Tasks

package com.example.async.service;
 
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
 
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
 
@Service
@Slf4j
public class ScheduledTaskService {
 
    /**
     * Fixed rate: chạy mỗi 60 giây bất kể thời gian thực thi trước
     */
    @Scheduled(fixedRate = 60000)
    public void cleanExpiredSessions() {
        log.info("Cleaning expired sessions at {}", now());
        // sessionRepository.deleteExpired();
    }
 
    /**
     * Fixed delay: chờ 30 giây SAU KHI thực thi trước hoàn thành
     */
    @Scheduled(fixedDelay = 30000)
    public void syncExternalData() {
        log.info("Syncing external data at {}", now());
        // externalApiClient.sync();
    }
 
    /**
     * Initial delay: chờ 10 giây trước lần thực thi đầu,
     * sau đó chạy mỗi 5 phút
     */
    @Scheduled(fixedRate = 300000, initialDelay = 10000)
    public void refreshCache() {
        log.info("Refreshing cache at {}", now());
        // cacheService.refreshAll();
    }
 
    /**
     * Cron expression: chạy lúc 9:00 sáng mỗi thứ Hai
     * Format: giây phút giờ ngày-trong-tháng tháng ngày-trong-tuần
     */
    @Scheduled(cron = "0 0 9 * * MON")
    public void generateWeeklyReport() {
        log.info("Generating weekly report at {}", now());
        // reportService.generateWeekly();
    }
 
    /**
     * Cron: chạy mỗi ngày lúc nửa đêm
     */
    @Scheduled(cron = "0 0 0 * * *")
    public void dailyCleanup() {
        log.info("Running daily cleanup at {}", now());
        // cleanupService.runDaily();
    }
 
    /**
     * Cron: chạy mỗi 15 phút trong giờ làm việc (Thứ 2-6, 9AM-5PM)
     */
    @Scheduled(cron = "0 */15 9-17 * * MON-FRI")
    public void checkHealthDuringBusinessHours() {
        log.info("Health check at {}", now());
        // healthService.check();
    }
 
    private String now() {
        return LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
    }
}

Tham Khảo Cron Expression

┌───────────── giây (0-59)
│ ┌───────────── phút (0-59)
│ │ ┌───────────── giờ (0-23)
│ │ │ ┌───────────── ngày trong tháng (1-31)
│ │ │ │ ┌───────────── tháng (1-12)
│ │ │ │ │ ┌───────────── ngày trong tuần (0-7, SUN-SAT)
│ │ │ │ │ │
* * * * * *

Các mẫu phổ biến:

Expression	Mô Tả
`0 0 * * * *`	Mỗi giờ
`0 /15 * * *`	Mỗi 15 phút
`0 0 9 * * MON-FRI`	Ngày làm việc lúc 9 sáng
`0 0 0 * * *`	Mỗi nửa đêm
`0 0 0 1 * *`	Ngày đầu mỗi tháng
`0 0 9 * * MON`	Mỗi thứ Hai lúc 9 sáng

Externalize Cron Expressions

Giữ lịch có thể cấu hình qua properties:

# application.yml
app:
  scheduling:
    weekly-report: "0 0 9 * * MON"
    daily-cleanup: "0 0 0 * * *"
    health-check: "0 */15 9-17 * * MON-FRI"

@Scheduled(cron = "${app.scheduling.weekly-report}")
public void generateWeeklyReport() {
    // Cron expression được tải từ cấu hình
}

Scheduling Có Điều Kiện

Tắt scheduling trong môi trường test:

@Configuration
@EnableScheduling
@ConditionalOnProperty(name = "app.scheduling.enabled", havingValue = "true", matchIfMissing = true)
public class SchedulingConfig {
}

# application-test.yml
app:
  scheduling:
    enabled: false

Xử Lý Lỗi Trong Phương Thức Async

Xử Lý Lỗi Trong Phương Thức @Async Trả Về void

Với phương thức @Async trả về void, exception không được xử lý sẽ được log nhưng mất âm thầm. Sử dụng exception handler tùy chỉnh:

@Configuration
@EnableAsync
@Slf4j
public class AsyncConfig implements AsyncConfigurer {
 
    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return (ex, method, params) -> {
            log.error("Unhandled async exception in {}.{}(): {}",
                method.getDeclaringClass().getSimpleName(),
                method.getName(),
                ex.getMessage(),
                ex
            );
 
            // Tùy chọn gửi cảnh báo, ghi vào dead letter queue, v.v.
            // alertService.sendAlert("Async failure: " + method.getName());
        };
    }
}

Xử Lý Lỗi Trong Phương Thức CompletableFuture

Với phương thức trả về CompletableFuture, xử lý lỗi tại nơi gọi:

@Async
public CompletableFuture<String> riskyOperation() {
    // Nếu throw exception, future hoàn thành với lỗi
    String result = externalApi.call();
    return CompletableFuture.completedFuture(result);
}
 
// Nơi gọi xử lý lỗi:
dataService.riskyOperation()
    .thenAccept(result -> log.info("Success: {}", result))
    .exceptionally(ex -> {
        log.error("Operation failed: {}", ex.getMessage());
        return null;
    });

Mẫu Retry Cho Thao Tác Async

package com.example.async.service;
 
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
 
import java.util.concurrent.CompletableFuture;
 
@Service
@Slf4j
public class ResilientNotificationService {
 
    private static final int MAX_RETRIES = 3;
    private static final long RETRY_DELAY_MS = 1000;
 
    @Async
    public CompletableFuture<Boolean> sendEmailWithRetry(String to, String subject, String body) {
        for (int attempt = 1; attempt <= MAX_RETRIES; attempt++) {
            try {
                log.info("Attempt {} to send email to {}", attempt, to);
                sendEmailInternal(to, subject, body);
                return CompletableFuture.completedFuture(true);
            } catch (Exception e) {
                log.warn("Attempt {} failed for {}: {}", attempt, to, e.getMessage());
                if (attempt < MAX_RETRIES) {
                    try {
                        Thread.sleep(RETRY_DELAY_MS * attempt); // Exponential backoff
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                        return CompletableFuture.completedFuture(false);
                    }
                }
            }
        }
        log.error("All {} attempts failed for {}", MAX_RETRIES, to);
        return CompletableFuture.completedFuture(false);
    }
 
    private void sendEmailInternal(String to, String subject, String body) {
        // Logic gửi email thực tế
        // mailSender.send(message);
        if (Math.random() < 0.3) {
            throw new RuntimeException("SMTP connection failed");
        }
    }
}

Dự Án Thực Hành: Dịch Vụ Thông Báo Nền

Hãy xây dựng một dịch vụ thông báo hoàn chỉnh minh họa xử lý async trong thực tế.

Domain Model

package com.example.async.model;
 
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
 
import java.time.LocalDateTime;
 
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class Notification {
    private Long id;
    private String recipient;
    private String channel;    // EMAIL, SMS, PUSH
    private String subject;
    private String message;
    private String status;     // PENDING, SENT, FAILED
    private LocalDateTime createdAt;
    private LocalDateTime sentAt;
    private int retryCount;
}

Notification Service

package com.example.async.service;
 
import com.example.async.model.Notification;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
 
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
 
@Service
@RequiredArgsConstructor
@Slf4j
public class BackgroundNotificationService {
 
    private final AtomicLong idGenerator = new AtomicLong(0);
    private final Map<Long, Notification> notificationStore = new ConcurrentHashMap<>();
 
    /**
     * Xếp hàng notification để gửi async
     */
    public Notification queueNotification(String recipient, String channel, String subject, String message) {
        Notification notification = Notification.builder()
            .id(idGenerator.incrementAndGet())
            .recipient(recipient)
            .channel(channel)
            .subject(subject)
            .message(message)
            .status("PENDING")
            .createdAt(LocalDateTime.now())
            .retryCount(0)
            .build();
 
        notificationStore.put(notification.getId(), notification);
        log.info("Queued notification #{} ({}) to {}", notification.getId(), channel, recipient);
 
        // Xử lý bất đồng bộ
        processNotificationAsync(notification);
        return notification;
    }
 
    @Async("notificationExecutor")
    public void processNotificationAsync(Notification notification) {
        try {
            log.info("Processing notification #{} on thread: {}",
                notification.getId(), Thread.currentThread().getName());
 
            switch (notification.getChannel()) {
                case "EMAIL" -> sendEmail(notification);
                case "SMS" -> sendSms(notification);
                case "PUSH" -> sendPush(notification);
                default -> throw new IllegalArgumentException("Unknown channel: " + notification.getChannel());
            }
 
            notification.setStatus("SENT");
            notification.setSentAt(LocalDateTime.now());
            log.info("Notification #{} sent successfully", notification.getId());
 
        } catch (Exception e) {
            notification.setRetryCount(notification.getRetryCount() + 1);
            if (notification.getRetryCount() >= 3) {
                notification.setStatus("FAILED");
                log.error("Notification #{} permanently failed after {} attempts",
                    notification.getId(), notification.getRetryCount());
            } else {
                notification.setStatus("PENDING");
                log.warn("Notification #{} failed (attempt {}), will retry",
                    notification.getId(), notification.getRetryCount());
            }
        }
    }
 
    /**
     * Retry notification thất bại mỗi 5 phút
     */
    @Scheduled(fixedRate = 300000)
    public void retryFailedNotifications() {
        List<Notification> pending = notificationStore.values().stream()
            .filter(n -> "PENDING".equals(n.getStatus()) && n.getRetryCount() > 0)
            .toList();
 
        if (!pending.isEmpty()) {
            log.info("Retrying {} failed notifications", pending.size());
            pending.forEach(this::processNotificationAsync);
        }
    }
 
    /**
     * Tạo báo cáo notification hàng ngày lúc nửa đêm
     */
    @Scheduled(cron = "0 0 0 * * *")
    public void generateDailyReport() {
        long total = notificationStore.size();
        long sent = notificationStore.values().stream()
            .filter(n -> "SENT".equals(n.getStatus())).count();
        long failed = notificationStore.values().stream()
            .filter(n -> "FAILED".equals(n.getStatus())).count();
        long pending = notificationStore.values().stream()
            .filter(n -> "PENDING".equals(n.getStatus())).count();
 
        log.info("Daily Notification Report - Total: {}, Sent: {}, Failed: {}, Pending: {}",
            total, sent, failed, pending);
    }
 
    public Map<String, Long> getStats() {
        long sent = notificationStore.values().stream()
            .filter(n -> "SENT".equals(n.getStatus())).count();
        long failed = notificationStore.values().stream()
            .filter(n -> "FAILED".equals(n.getStatus())).count();
        long pending = notificationStore.values().stream()
            .filter(n -> "PENDING".equals(n.getStatus())).count();
 
        return Map.of("total", (long) notificationStore.size(),
            "sent", sent, "failed", failed, "pending", pending);
    }
 
    public Notification getNotification(Long id) {
        return notificationStore.get(id);
    }
 
    private void sendEmail(Notification notification) throws InterruptedException {
        Thread.sleep(1500); // Mô phỏng gọi SMTP
    }
 
    private void sendSms(Notification notification) throws InterruptedException {
        Thread.sleep(800); // Mô phỏng SMS API
    }
 
    private void sendPush(Notification notification) throws InterruptedException {
        Thread.sleep(300); // Mô phỏng push notification
    }
}

Notification Controller

package com.example.async.controller;
 
import com.example.async.model.Notification;
import com.example.async.service.BackgroundNotificationService;
import lombok.RequiredArgsConstructor;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
 
import java.util.Map;
 
@RestController
@RequestMapping("/api/notifications")
@RequiredArgsConstructor
public class NotificationController {
 
    private final BackgroundNotificationService notificationService;
 
    @PostMapping
    public ResponseEntity<Notification> sendNotification(@RequestBody Map<String, String> request) {
        Notification notification = notificationService.queueNotification(
            request.get("recipient"),
            request.get("channel"),
            request.get("subject"),
            request.get("message")
        );
        return ResponseEntity.accepted().body(notification);
    }
 
    @GetMapping("/{id}")
    public ResponseEntity<Notification> getNotification(@PathVariable Long id) {
        Notification notification = notificationService.getNotification(id);
        if (notification == null) {
            return ResponseEntity.notFound().build();
        }
        return ResponseEntity.ok(notification);
    }
 
    @GetMapping("/stats")
    public ResponseEntity<Map<String, Long>> getStats() {
        return ResponseEntity.ok(notificationService.getStats());
    }
}

Kiểm Tra Dịch Vụ Notification

# Gửi email notification
curl -X POST http://localhost:8080/api/notifications \
  -H "Content-Type: application/json" \
  -d '{"recipient":"[email protected]","channel":"EMAIL","subject":"Hello","message":"Welcome!"}'
 
# Gửi SMS notification
curl -X POST http://localhost:8080/api/notifications \
  -H "Content-Type: application/json" \
  -d '{"recipient":"+1234567890","channel":"SMS","subject":"Alert","message":"Your order shipped"}'
 
# Kiểm tra trạng thái notification
curl http://localhost:8080/api/notifications/1
 
# Xem thống kê
curl http://localhost:8080/api/notifications/stats

Kiểm Thử Code Async

Kiểm thử phương thức async cần xử lý đặc biệt vì assertions có thể chạy trước khi tác vụ async hoàn thành.

Unit Testing Phương Thức @Async

package com.example.async.service;
 
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
 
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
 
import static org.assertj.core.api.Assertions.assertThat;
 
@SpringBootTest
class DataAggregationServiceTest {
 
    @Autowired
    private DataAggregationService dataService;
 
    @Test
    void shouldFetchUserProfileAsync() throws Exception {
        // When
        CompletableFuture<String> future = dataService.fetchUserProfile(1L);
 
        // Then - chờ kết quả async
        String result = future.get(5, TimeUnit.SECONDS);
        assertThat(result).contains("User-1");
    }
 
    @Test
    void shouldRunTasksInParallel() throws Exception {
        long start = System.currentTimeMillis();
 
        CompletableFuture<String> profile = dataService.fetchUserProfile(1L);
        CompletableFuture<String> orders = dataService.fetchUserOrders(1L);
        CompletableFuture<String> recs = dataService.fetchUserRecommendations(1L);
 
        CompletableFuture.allOf(profile, orders, recs).get(10, TimeUnit.SECONDS);
 
        long duration = System.currentTimeMillis() - start;
 
        // Phải chạy song song (~2000ms), không tuần tự (~4500ms)
        assertThat(duration).isLessThan(3500);
 
        assertThat(profile.get()).contains("User-1");
        assertThat(orders.get()).contains("5 orders");
        assertThat(recs.get()).contains("recommended");
    }
}

Kiểm Thử Scheduled Tasks

package com.example.async.service;
 
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.mock.mockito.SpyBean;
import org.awaitility.Awaitility;
 
import java.util.concurrent.TimeUnit;
 
import static org.mockito.Mockito.atLeast;
import static org.mockito.Mockito.verify;
 
@SpringBootTest(properties = {
    "app.scheduling.enabled=true"
})
class ScheduledTaskServiceTest {
 
    @SpyBean
    private ScheduledTaskService scheduledTaskService;
 
    @Test
    void shouldExecuteScheduledTask() {
        // Sử dụng Awaitility để chờ scheduled task
        Awaitility.await()
            .atMost(70, TimeUnit.SECONDS)
            .untilAsserted(() ->
                verify(scheduledTaskService, atLeast(1)).cleanExpiredSessions()
            );
    }
}

Thêm Awaitility vào test dependencies để kiểm thử async dễ dàng hơn:
<dependency>
    <groupId>org.awaitility</groupId>
    <artifactId>awaitility</artifactId>
    <scope>test</scope>
</dependency>

Best Practices Cho Production

1. Luôn Cấu Hình Thread Pool

Không bao giờ dùng SimpleAsyncTaskExecutor mặc định trong production. Nó tạo thread không giới hạn có thể làm crash ứng dụng khi tải cao.

2. Đặt Tên Thread

Tiền tố tên thread giúp debug dễ dàng hơn rất nhiều:

executor.setThreadNamePrefix("email-");  // Hiển thị email-1, email-2, v.v.

3. Xử Lý Shutdown Mượt Mà

executor.setWaitForTasksToCompleteOnShutdown(true);
executor.setAwaitTerminationSeconds(30);

4. Giám Sát Metrics Thread Pool

Expose thống kê thread pool qua Actuator:

@Bean
public ThreadPoolTaskExecutor taskExecutor(MeterRegistry registry) {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(5);
    executor.setMaxPoolSize(10);
    executor.setThreadNamePrefix("async-");
    executor.initialize();
 
    // Đăng ký metrics
    registry.gauge("async.pool.size", executor, ThreadPoolTaskExecutor::getPoolSize);
    registry.gauge("async.pool.active", executor, ThreadPoolTaskExecutor::getActiveCount);
    registry.gauge("async.queue.size", executor, e -> e.getThreadPoolExecutor().getQueue().size());
 
    return executor;
}

5. Tránh Các Lỗi Phổ Biến

Gọi @Async từ cùng class:

// SAI - @Async bị bỏ qua (không có proxy)
@Service
public class MyService {
    public void doWork() {
        asyncMethod(); // Chạy đồng bộ!
    }
 
    @Async
    public void asyncMethod() { }
}
 
// ĐÚNG - Gọi từ bean khác
@Service
public class CallerService {
    private final AsyncService asyncService;
 
    public void doWork() {
        asyncService.asyncMethod(); // Chạy bất đồng bộ
    }
}

Thiếu @EnableAsync:

// @Async không có tác dụng nếu thiếu @EnableAsync trên class @Configuration

Không xử lý InterruptedException đúng cách:

// ĐÚNG
catch (InterruptedException e) {
    Thread.currentThread().interrupt(); // Khôi phục interrupt flag
    throw new RuntimeException("Task interrupted", e);
}

Tóm Tắt và Điểm Chính

Bạn đã học cách xây dựng các ứng dụng bất đồng bộ, non-blocking với Spring Boot.

Những Gì Đã Học:

✅ @Async cho tác vụ nền fire-and-forget
✅ CompletableFuture cho phương thức async có giá trị trả về
✅ Cấu hình thread pool tùy chỉnh cho production
✅ Nhiều thread pool cho cô lập workload
✅ @Scheduled cho fixed-rate, fixed-delay, và cron tasks
✅ Xử lý lỗi và mẫu retry
✅ Xây dựng dịch vụ thông báo nền
✅ Kiểm thử code async và scheduled

Quy Tắc Chính:

Luôn cấu hình ThreadPoolTaskExecutor phù hợp cho production
Gọi phương thức @Async từ bean khác (hạn chế proxy)
Sử dụng CompletableFuture khi cần kết quả từ thao tác async
Xử lý InterruptedException đúng cách — khôi phục interrupt flag
Externalize cron expressions qua properties để linh hoạt

Bước Tiếp Theo:

Thực hành: Thêm async email notifications vào dự án hiện có
Khám phá: Tìm hiểu Spring Events (@EventListener) cho xử lý async decoupled
Nâng cấp: Học message queues (RabbitMQ, Kafka) cho xử lý async phân tán
Giám sát: Thêm thread pool metrics vào Actuator dashboards

Câu Hỏi Thường Gặp

Q: Sự khác biệt giữa fixedRate và fixedDelay là gì?
A: fixedRate kích hoạt ở khoảng thời gian cố định bất kể thời gian thực thi. Nếu tác vụ mất 3 giây và rate là 5 giây, lần chạy tiếp theo bắt đầu 2 giây sau khi lần trước kết thúc. fixedDelay chờ thời gian được chỉ định sau khi lần thực thi trước hoàn thành — nên delay 5 giây luôn nghĩa là 5 giây idle giữa các lần chạy.

Q: Có thể dùng @Async với @Transactional không?
A: Có, nhưng lưu ý phương thức async chạy trong thread riêng với transaction context riêng. Transaction của phương thức gọi sẽ không truyền sang phương thức async. Nếu cần transactional behavior, đánh dấu @Transactional trên chính phương thức async.

Q: Nên cấu hình bao nhiêu thread?
A: Cho tác vụ I/O-bound (HTTP calls, email, database), bắt đầu với 2× CPU cores cho core size và 4× cho max. Cho tác vụ CPU-bound, đặt core size bằng số CPU. Giám sát dưới tải và điều chỉnh.

Q: Điều gì xảy ra khi thread pool đầy?
A: Mặc định, tác vụ đi vào queue. Khi queue cũng đầy, RejectedExecutionHandler được gọi. Các tùy chọn bao gồm: AbortPolicy (throw exception), CallerRunsPolicy (chạy trên calling thread), DiscardPolicy (bỏ âm thầm), hoặc handler tùy chỉnh.

Q: Nên dùng @Async hay message queue như RabbitMQ?
A: Dùng @Async cho tác vụ nền đơn giản trong cùng process. Dùng message queue khi cần: persistence tác vụ qua restart, xử lý phân tán trên nhiều server, delivery đảm bảo, hoặc routing phức tạp. Message queues thêm độ phức tạp vận hành nhưng cung cấp đảm bảo mạnh hơn.

Tài Liệu Tham Khảo

Documentation:

Hướng Dẫn Liên Quan:

Bắt Đầu với Spring Boot - Nền tảng
Tích Hợp Database với JPA - Lớp dữ liệu
Caching với Redis - Tối ưu hiệu suất
Hướng Dẫn Testing với JUnit & Mockito - Mẫu testing
Lộ Trình Học Spring Boot - Lộ trình học đầy đủ

Chúc bạn code vui! Nếu hướng dẫn này hữu ích, hãy xem Lộ Trình Học Spring Boot để tìm hiểu thêm nhiều chủ đề khác.