java concurrency

Backend bây giờ phải xử lý đồng thời (concurrency) rất nhiều việc mà vẫn giữ độ trễ thấp. Java 21 cùng Spring Boot 3.x thay đổi khá nhiều cách mình tổ chức code cho chuyện này.

Bài này là dạng ghi chép tra cứu: các pattern hay dùng, vài best practice và mấy cái "hố" (pitfall) dễ sập khi làm concurrency kiểu mới của Java. Một số API ở đây (ScopedValue, Structured Concurrency) còn là preview ở Java 21 và đã thay đổi ở các bản sau, mình ghi chú ngay chỗ đó.

# virtual threads (project loom) trong java 21

# khái niệm

Virtual Threads là thread nhẹ do JVM quản lý. Chúng được "gắn" (mount) lên một carrier thread (platform thread thật) để chạy, và khi gặp thao tác blocking (I/O, sleep, chờ lock...) thì JVM gỡ (unmount) ra, cất stack vào heap, nhường carrier cho virtual thread khác. Platform thread thì ánh xạ 1:1 với OS thread, mỗi cái giữ stack cỡ 1MB; virtual thread chỉ tốn vài trăm byte tới vài KB lúc nghỉ, nên tạo hàng triệu cái vẫn được.

// Tạo virtual thread trực tiếp
Thread.startVirtualThread(() -> doWork());
 
// Factory pattern — có naming cho debug/logging
ThreadFactory factory = Thread.ofVirtual()
       .name("my-vt-", 0)  // tên prefix + counter
       .factory();
 
// Executor pattern — mỗi task 1 virtual thread
ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
 
// Scheduled executor với virtual threads
ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(
       Thread.ofVirtual().name("cache-cleanup").factory()
);

# khi nào dùng virtual threads

# pitfalls

• Pinning: trên Java 21-23, nếu virtual thread blocking khi đang ở trong khối synchronized (hoặc trong native method/JNI) thì nó không unmount được, carrier thread bị "ghim" (pin) lại. Vài chỗ như vậy thì không sao, nhưng nhiều request cùng bị ghim là hết carrier, cả ứng dụng đứng hình. Cách xử lý trên các bản này: thay bằng ReentrantLock, và bật -Djdk.tracePinnedThreads=full để tìm chỗ bị ghim. Từ Java 24 (JEP 491), synchronized không còn gây pinning nữa; native method và một số trường hợp hiếm thì vẫn còn.
• ThreadLocal: vẫn dùng được, nhưng hàng triệu virtual thread, mỗi cái giữ một bản ThreadLocal riêng, thì memory phình rất nhanh. Đặc biệt đừng dùng ThreadLocal để cache object đắt (kiểu SimpleDateFormat) vì virtual thread không được tái sử dụng. Giải pháp: ưu tiên ScopedValue cho dữ liệu ngữ cảnh.
• Pool sizing không có nghĩa: với Virtual Threads, khái niệm "pool" gần như không còn. newVirtualThreadPerTaskExecutor() tạo một thread riêng cho mỗi task và không giới hạn số lượng. Đừng gom virtual thread vào pool.
• Connection pool exhaustion: ảo hoá thread không có nghĩa là ảo hoá luôn tài nguyên thật. Hàng triệu virtual thread vẫn tranh nhau vài chục connection DB, hoặc dội thẳng vào API bên ngoài. Giải pháp: giới hạn bằng Semaphore (xem phần dưới).

# spring boot integration

# application.yml — enable globally
spring:
  threads:
    virtual:
      enabled: true

Khi enabled, Spring Boot 3.2+ tự động:

• Tomcat sử dụng virtual threads cho request handling
• @Async methods chạy trên virtual threads (executor mặc định applicationTaskExecutor thành SimpleAsyncTaskExecutor dùng virtual thread)
• RabbitMQ listener threads là virtual (Kafka listener và @Scheduled cũng vậy)

Lưu ý SimpleAsyncTaskExecutor không có queue hay giới hạn số thread: mỗi task một virtual thread. Nếu @Async gọi DB thì vẫn phải tự chặn đồng thời như phần Semaphore bên dưới.

# ScopedValue (preview ở java 21, chính thức ở java 25): thay thế ThreadLocal

# vấn đề với ThreadLocal

• Mutable: ai cũng set() được ở bất kỳ đâu, khó lần ra giá trị bị đổi lúc nào. Chia sẻ object mutable giữa luồng cha và con còn dễ gây race condition.
• Memory leak: phải tự remove(). Với thread pool, quên dọn là request sau đọc nhầm dữ liệu của request trước.
• Chi phí kế thừa: InheritableThreadLocal copy toàn bộ giá trị sang mỗi luồng con lúc tạo, tạo càng nhiều virtual thread thì càng tốn.

ScopedValue giải quyết bằng cách gắn giá trị vào một phạm vi thực thi: bind ở đầu khối run(), hết khối là tự mất, bên trong không ai sửa được. Nó là preview ở Java 21-24 (cần --enable-preview, JEP 446) và thành tính năng chính thức ở Java 25 (JEP 506).

# ScopedValue pattern

// Khai báo — Bất biến (Immutable) và gắn chặt với một Scope nhất định
public static final ScopedValue<String> REQUEST_ID = ScopedValue.newInstance();
 
// Ràng buộc (Bind) và Thực thi
ScopedValue.where(REQUEST_ID, "trace-123").run(() -> {
   // REQUEST_ID.get() = "trace-123" trong scope này
   doWork();
});
 
// Read — throws NoSuchElementException nếu chưa bound
String id = REQUEST_ID.get();
 
// Truy xuất linh hoạt (Kiểm tra kỹ trước khi đọc để tránh NoSuchElementException)
if (REQUEST_ID.isBound()) {
   String safeId = REQUEST_ID.get();
}
 
// Hoặc lấy giá trị mặc định
String idOrDefault = REQUEST_ID.orElse("unknown");

# so sánh

# context propagation across threads

Khi chuyển việc sang luồng khác (Platform hay Virtual đều vậy), ngữ cảnh của luồng gọi như TraceId, MDC (logging), Security Context hay Locale sẽ không đi theo, vì chúng đều nằm trong ThreadLocal. ScopedValue cũng vậy: nó chỉ tự truyền xuống subtask của StructuredTaskScope, còn task gửi vào một executor bình thường thì không. Trong hệ thống phân tán, log mất traceId là lúc debug khổ nhất.

# TaskDecorator pattern

public class ContextPropagatingDecorator implements TaskDecorator {
   @Override
   public Runnable decorate(Runnable runnable) {
       // 1. Chụp lại (Capture) toàn bộ context từ luồng gọi (Caller Thread)
       String traceId = RequestContext.getRequestId();
       Map<String, String> mdc = MDC.getCopyOfContextMap();
       Locale locale = LocaleContextHolder.getLocale();
 
       // 2. Đóng gói vào một Runnable mới
       return () -> {
           // 3. Phục hồi (Restore) context tại luồng thực thi (Worker Thread)
           if (mdc != null) MDC.setContextMap(mdc);
           LocaleContextHolder.setLocale(locale);
           try {
               ScopedValue.where(RequestContext.REQUEST_ID, traceId)
                   .run(runnable);
           } finally {
                // 4. Dọn dẹp sạch sẽ
               MDC.clear();
               LocaleContextHolder.resetLocaleContext();
           }
       };
   }
}

Spring Framework 6.1+ có sẵn ContextPropagatingTaskDecorator (dựa trên thư viện Micrometer Context Propagation) làm đúng việc này cho MDC, trace, Security Context... đã đăng ký accessor. Còn Spring Boot 3.x tự gắn bean TaskDecorator duy nhất trong context vào executor mà nó auto-configure. Tự viết như trên chỉ cần khi có ngữ cảnh riêng của bạn.

# BeanPostProcessor cho auto-decoration

Inject TaskDecorator vào tất cả executor beans tự động, để không phải cấu hình thủ công cho từng Executor:

@Component
public class ExecutorContextInjector implements BeanPostProcessor {
   @Override
   public Object postProcessBeforeInitialization(Object bean, String beanName) {
       // Làm ở BEFORE: Spring 6.0 chỉ đọc taskDecorator lúc initialize()
       if (bean instanceof ThreadPoolTaskExecutor executor) {
           // Lưu ý: ghi đè decorator đã set trước đó (nếu có)
           executor.setTaskDecorator(new ContextPropagatingDecorator());
       }
       return bean;
   }
 
   @Override
   public Object postProcessAfterInitialization(Object bean, String beanName) {
       // Các AsyncTaskExecutor khác: bọc lại, wrapper phải implement AsyncTaskExecutor
       if (bean instanceof AsyncTaskExecutor asyncExecutor
               && !(bean instanceof ThreadPoolTaskExecutor)) {
           return new TracePropagatingWrapper(asyncExecutor);
       }
       return bean;
   }
}

Note: không thay bean bằng lambda/wrapper khác type, vì sẽ gây BeanNotOfRequiredTypeException. Phải giữ nguyên type gốc của bean. Ngay cả wrapper implement AsyncTaskExecutor thì chỗ nào inject theo class cụ thể (ThreadPoolTaskExecutor) vẫn lỗi, nên ở trên ThreadPoolTaskExecutor chỉ được set decorator chứ không bọc. Set decorator nên làm ở postProcessBeforeInitialization: tới AfterInitialization thì initialize() đã chạy xong, và ở Spring 6.0 (Boot 3.0/3.1) decorator set sau lúc đó bị bỏ qua. Từ Spring 6.1 executor đọc decorator lúc chạy nên set muộn vẫn ăn, nhưng làm ở Before thì đúng với mọi bản.

# @Async pattern

Cách đơn giản nhất để chạy bất đồng bộ trong Spring, nhưng cần theo vài quy tắc để khỏi dính các bẫy phổ biến.

@Service
public class NotificationService {
 
   // Fire-and-forget
   @Async
   public void sendEmail(String to, String body) {
       // runs on async executor
   }
 
   // Return result
   @Async
   public CompletableFuture<Report> generateReport(String id) {
       Report report = heavyComputation(id);
       return CompletableFuture.completedFuture(report);
   }
 
   // Chỉ định executor cụ thể
   @Async("auditExecutor")
   public void audit(AuditEvent event) {
       // runs on named executor
   }
}

# vài quy tắc với @async

• Method nên public (Spring dùng proxy-based AOP; method private chắc chắn không được proxy chặn).
• Không gọi hàm @Async từ một hàm khác trong cùng một class (self-invocation không đi qua proxy nên chạy đồng bộ như hàm thường).
• Exception trong hàm void không về tới caller. Mặc định Spring chỉ log nó qua SimpleAsyncUncaughtExceptionHandler; muốn xử lý riêng (alert, retry) thì implement AsyncConfigurer.getAsyncUncaughtExceptionHandler().
• Exception trong CompletableFuture sẽ propagate khi .join() / .get()
• Đánh dấu @EnableAsync ở tầng cấu hình một lần (và chỉ một lần).

# custom Executor

@Configuration
public class AsyncConfig {
 
   @Bean(name = "auditExecutor")
   public AsyncTaskExecutor auditExecutor() {
       ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
       executor.setCorePoolSize(2);
       executor.setMaxPoolSize(5);
       executor.setQueueCapacity(100);
       executor.setThreadNamePrefix("audit-");
       executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
       // Không cần gọi initialize(): Spring tự gọi qua afterPropertiesSet()
       return executor;
   }
}

Với cấu hình này, thread chỉ tăng quá 2 khi queue 100 chỗ đã đầy (đó là cách ThreadPoolExecutor hoạt động: core → queue → max). CallerRunsPolicy nghĩa là khi pool và queue đều đầy, chính thread gọi audit() sẽ tự chạy task, tức là @Async thành đồng bộ đúng lúc hệ thống đang quá tải. Đó là cơ chế backpressure có chủ đích, nhưng nên biết để khỏi bất ngờ.

Thêm một bẫy với Spring Boot: mặc định, khai báo bất kỳ bean Executor nào là auto-configured applicationTaskExecutor lùi lại không tạo nữa. Khi đó các @Async không chỉ định tên có thể rơi vào bean executor duy nhất bạn vừa khai báo, hoặc SimpleAsyncTaskExecutor nếu có nhiều bean. Chỉ định tên executor rõ ràng cho từng @Async là an toàn nhất.

# CompletableFuture patterns

# fan-out / fan-in - phân tán & gom tụ

// Chạy nhiều tasks song song, đợi tất cả
CompletableFuture<List<Label>> labelsFuture = CompletableFuture.supplyAsync(
   () -> labelService.verify(metadata), executor);
CompletableFuture<Boolean> mimeFuture = CompletableFuture.supplyAsync(
   () -> mimeService.check(file), executor);
 
// Đợi tất cả hoàn thành
CompletableFuture.allOf(labelsFuture, mimeFuture).join();
 
// Lấy kết quả (đã complete)
List<Label> labels = labelsFuture.join();
Boolean mimeOk = mimeFuture.join();

# chaining và xử lý lỗi

CompletableFuture.supplyAsync(() -> fetchData(), executor)
   .thenApply(data -> transform(data))
   .thenAccept(result -> save(result))
   .exceptionally(ex -> {
       log.error("Pipeline failed", ex);
       return null;
   });

# timeout handling

CompletableFuture<Result> future = CompletableFuture.supplyAsync(() -> slowOp(), executor);
Result result = future.orTimeout(5, TimeUnit.SECONDS)  // Java 9+
   .exceptionally(ex -> fallback())
   .join();

orTimeout chỉ làm future hoàn thành với TimeoutException; nó không ngắt slowOp() đang chạy (CompletableFuture không interrupt thread). Task vẫn chiếm thread tới khi xong, nên bản thân slowOp() cũng cần timeout riêng (timeout của HTTP client, JDBC query timeout). Muốn trả giá trị mặc định thay vì exception thì có completeOnTimeout(value, 5, SECONDS).

# gotchas

• .join() ném CompletionException (unchecked), còn .get() ném ExecutionException (checked)
• Không truyền executor thì dùng ForkJoinPool.commonPool() (platform threads, limited)
• allOf() chờ tất cả xong rồi mới hoàn thành; nếu có task fail thì allOf().join() ném exception (các task khác vẫn chạy hết, không bị huỷ). Dùng handle() trên từng future nếu muốn partial results
• Virtual thread + CompletableFuture: luôn truyền explicit executor

# semaphore-based concurrency control

Như đã nói ở trên, virtual thread tạo ra không giới hạn có thể làm sập API bên ngoài hoặc database. Semaphore là chốt chặn khá hiệu quả: thread nào không lấy được permit thì đứng chờ, và với virtual thread thì việc chờ này gần như không tốn gì.

@Component
public class ThrottledExecutor {
   private final ExecutorService executor;
   private final Semaphore semaphore;
 
   public ThrottledExecutor(
       @Value("${app.executor.max-concurrent-tasks:50}") int maxConcurrent,
       @Value("${app.executor.use-virtual-thread:true}") boolean useVirtual
   ) {
       this.executor = useVirtual
           ? Executors.newVirtualThreadPerTaskExecutor()
           : new ThreadPoolExecutor(4, 50, 60, SECONDS, new LinkedBlockingQueue<>(500));
 
       // Semaphore = null nếu unlimited
       this.semaphore = maxConcurrent > 0 ? new Semaphore(maxConcurrent) : null;
   }
 
   public <T> CompletableFuture<T> submit(Callable<T> task) {
       return CompletableFuture.supplyAsync(() -> {
           acquire();
           try {
               return task.call();
           } catch (Exception e) {
               throw new RuntimeException(e);
           } finally {
               release();
           }
       }, executor);
   }
 
   private void acquire() {
       if (semaphore != null) {
           try { semaphore.acquire(); }
           catch (InterruptedException e) {
               Thread.currentThread().interrupt();
               throw new RuntimeException("Interrupted waiting for permit", e);
           }
       }
   }
 
   private void release() {
       if (semaphore != null) semaphore.release();
   }
}

# ApplicationEvent + async multicaster

Tách rời logic nghiệp vụ (decoupling) bằng mô hình Publisher - Listener. Từ Spring 4.2, event không bắt buộc phải kế thừa ApplicationEvent, một record bình thường cũng publish được.

# pattern: async event handling

// Event class
public class OrderCreatedEvent extends ApplicationEvent {
   private final Order order;
   public OrderCreatedEvent(Object source, Order order) {
       super(source);
       this.order = order;
   }
}
 
// Publisher
@Service
public class OrderService {
   private final ApplicationEventPublisher publisher;
 
   public void createOrder(OrderRequest req) {
       Order order = save(req);
       publisher.publishEvent(new OrderCreatedEvent(this, order));
   }
}
 
// Listener — runs async nhờ multicaster config
@Component
public class NotificationListener {
   @EventListener
   public void onOrderCreated(OrderCreatedEvent event) {
       // chạy trên virtual thread (nếu multicaster dùng VT executor)
       sendNotification(event.getOrder());
   }
}

# multicaster config cho virtual threads

Để các Listener chạy Async thật sự trên Virtual Threads, cần chỉnh Multicaster:

@Configuration
public class EventConfig {
   @Bean(name = "applicationEventMulticaster")
   public ApplicationEventMulticaster multicaster() {
       SimpleApplicationEventMulticaster m = new SimpleApplicationEventMulticaster();
 
       ExecutorService vtExecutor = Executors.newThreadPerTaskExecutor(
           Thread.ofVirtual().name("event-vt-", 0).factory()
       );
 
       // Wrap với context propagation
       TaskExecutorAdapter adapter = new TaskExecutorAdapter(vtExecutor);
       TaskDecorator decorator = new ContextPropagatingDecorator();
       m.setTaskExecutor(task -> adapter.execute(decorator.decorate(task)));
 
       return m;
   }
}

Lưu ý: Multicaster async nghĩa là gần như tất cả listener chạy async, kể cả listener cho event nội bộ của Spring, exception của listener cũng không về tới publisher. Từ Spring 6.1, listener nào trả supportsAsyncExecution() == false (trong đó có @TransactionalEventListener) thì vẫn được gọi đồng bộ; trước 6.1, @TransactionalEventListener chạy trên thread khác sẽ không thấy transaction và event bị bỏ qua. Nếu chỉ vài listener cần async, cách ít rủi ro hơn là giữ multicaster mặc định và gắn @Async lên đúng listener đó.

# deferredresult: non-blocking controller response

# pattern

@RestController
public class ReportController {
 
   @GetMapping("/reports/{id}")
   public DeferredResult<ResponseEntity<?>> getReport(@PathVariable String id) {
       DeferredResult<ResponseEntity<?>> result = new DeferredResult<>(5000L);
 
       // Capture context từ request thread
       String traceId = RequestContext.getRequestId();
       Map<String, String> mdc = MDC.getCopyOfContextMap();
 
       Thread.startVirtualThread(() -> {
           if (mdc != null) MDC.setContextMap(mdc);
           RequestContext.runWith(traceId, () -> {
               try {
                   Report report = reportService.generate(id);
                   result.setResult(ResponseEntity.ok(report));
               } catch (Exception e) {
                   result.setErrorResult(e);
               } finally {
                   MDC.clear();
               }
           });
       });
 
       result.onTimeout(() ->
           result.setResult(ResponseEntity.status(408).body("Timeout")));
 
       return result;
   }
}

# khi nào dùng DeferredResult

• Request cần xử lý lâu, không muốn giữ Tomcat thread trong suốt thời gian đó
• Muốn free Tomcat thread sớm cho request khác
• Response phụ thuộc vào external event (callback, message)

# so sánh với virtual threads trên tomcat

Nếu spring.threads.virtual.enabled=true thì Tomcat đã dùng virtual threads, DeferredResult ít cần hơn vì blocking trên virtual thread không tốn OS thread. Nó vẫn hữu ích khi cần explicit timeout control hoặc event-driven response. Trong ví dụ trên, timeout 5 giây chỉ cắt response, còn reportService.generate(id) vẫn chạy tiếp trên virtual thread cho tới khi xong.

# thread safety trong spring beans

# singleton scope (default)

Mặc định Spring Beans có scope Singleton (dùng chung cho tất cả các request threads).

@Service
public class UserService {
   // SAFE: final, immutable reference
   private final UserRepository repo;
 
   // UNSAFE: mutable shared state
   private int counter = 0; // ← race condition
 
   // SAFE: thread-safe collection
   private final AtomicInteger safeCounter = new AtomicInteger(0);
 
   // UNSAFE: non-thread-safe collection
   private final List<String> cache = new ArrayList<>(); // ← concurrent modification
 
   // SAFE: thread-safe collection
   private final List<String> safeCache = new CopyOnWriteArrayList<>();
}

# quy tắc

• Không lưu mutable state trong singleton bean
• Request-scoped data thì dùng method parameters, ScopedValue, hoặc RequestScope bean
• Cần shared mutable state thì dùng Atomic*, ConcurrentHashMap, hoặc explicit locking
• @Scope("prototype") tạo instance mới mỗi lần inject, không phải mỗi request

# structured concurrency (preview, java 21+)

# concept

Một mô hình xử lý đa luồng có tổ chức: gom các task có liên quan vào một nhóm, huỷ cả nhóm nếu một nhánh thất bại, và vòng đời được quản lý rõ ràng. Giống như cấu trúc try của code tuần tự: task con không thể sống lâu hơn khối code đã tạo ra nó. Code dưới đây là API ở Java 21-24 (JEP 453).

// Requires: --enable-preview
try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
   Subtask<User> userTask = scope.fork(() -> fetchUser(id));
   Subtask<List<Order>> ordersTask = scope.fork(() -> fetchOrders(id));
 
   scope.join();           // Block chờ toàn bộ nhánh con
   scope.throwIfFailed();  // Quăng exception ngay nếu có bất kỳ nhánh nào ngã ngựa
 
   User user = userTask.get();
   List<Order> orders = ordersTask.get();
   return new UserProfile(user, orders);
}
// Nếu fetchUser() fail → fetchOrders() bị cancel tự động

Điểm hay ở đây: nếu fetchUser() sập, fetchOrders() bị interrupt ngay, không chạy phí tới hết. Thread dump cũng hiện rõ quan hệ cha-con giữa các task.

# khi nào dùng

• Fan-out tasks có quan hệ (tất cả thành công hoặc tất cả fail)
• Thay thế CompletableFuture.allOf() khi cần proper cancellation
• Task hierarchy có parent-child relationship

# lưu ý

• Vẫn là preview (Java 21 tới 25), cần --enable-preview
• API đã đổi ở Java 25 (JEP 505): không còn new StructuredTaskScope.ShutdownOnFailure(), thay bằng StructuredTaskScope.open() với Joiner, và join() tự ném exception khi có subtask lỗi. Code viết cho 21 sẽ không biên dịch trên 25
• Spring chưa integrate native, nên dùng thủ công trong service layer

# common anti-patterns

# configuration properties

# Virtual threads
spring:
  threads:
    virtual:
      enabled: true
 
# Custom executor tuning
app:
  executor:
    use-virtual-thread: true
    max-concurrent-tasks: 50 # Semaphore permits (0 = unlimited)
    pool-size: 4 # fallback platform thread pool
    max-pool-size: 50
    queue-capacity: 500
    ttl: 60 # keep-alive seconds

# decision matrix

Chỉ là những ghi chép cá nhân với hy vọng mang lại chút giá trị. Nếu thấy hữu ích, đừng ngại chia sẻ cho bạn bè & đồng nghiệp nhé!

Happy coding 😎 👍🏻 🚀 🔥.