Java concurrency

// 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()
);

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

// 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 id = REQUEST_ID.get();
}

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();
           }
       };
   }
}

@Component
public class ExecutorContextInjector implements BeanPostProcessor {
   @Override
   public Object postProcessAfterInitialization(Object bean, String beanName) {
       if (bean instanceof ThreadPoolTaskExecutor executor) {
           // Compose với existing decorator, không override
           executor.setTaskDecorator(new ContextPropagatingDecorator());
       }
       if (bean instanceof AsyncTaskExecutor asyncExecutor) {
           return new TracePropagatingWrapper(asyncExecutor);
       }
       return bean;
   }
}

@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
   }
}

@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.setRejectionHandler(new CallerRunsPolicy());
       executor.initialize();
       return executor;
   }
}

// 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();

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

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

@Component
public class ThrottledExecutor {
   private final ExecutorService executor;
   private final Semaphore semaphore;
 
   public ThrottledExecutor(
       @Value("${max.concurrent:50}") int maxConcurrent,
       @Value("${use.virtual: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();
   }
}

// 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());
   }
}

@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;
   }
}

@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;
   }
}

@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<>();
}

// 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

# 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