Tất cả về Thread trong Java

// Single-threaded server - xử lý tuần tự
while (true) {
    Request req = acceptConnection();  // nhận request
    processOrder(req);                 // 250ms+ mỗi request
    // Request tiếp theo phải CHỜ request hiện tại xong
}

// Cách 1: Subclass Thread
// Vấn đề: không thể subclass thêm class khác, khó test, tightly coupled
class OrderProcessor extends Thread {
    @Override
    public void run() { processOrder(); }
}
new OrderProcessor().start();

// Cách 2: Runnable - tách biệt task khỏi cơ chế chạy
Runnable task = () -> processOrder();
new Thread(task).start();

ExecutorService executor = Executors.newFixedThreadPool(10);

// Callable: trả về kết quả, có thể throw Exception
Callable<ProductData> task = () -> {
    return externalApiClient.fetchProduct(productId); // có thể throw IOException
};

Future<ProductData> future = executor.submit(task);

// Làm việc khác trong khi task chạy async
doOtherWork();

// Lấy kết quả, block nếu chưa xong
try {
    ProductData data = future.get(5, TimeUnit.SECONDS); // timeout tránh block mãi
} catch (TimeoutException e) {
    future.cancel(true); // interrupt task nếu timeout
    throw new ServiceUnavailableException("External API timeout");
} catch (ExecutionException e) {
    throw new RuntimeException("Task failed", e.getCause());
}

// Bug: get() không timeout
ProductData data = future.get(); // block mãi nếu external API treo
// HTTP request của user cũng block mãi → timeout từ load balancer
// Thread bị giữ trong pool → pool exhaustion → toàn bộ service treo

ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(4);

// Chạy một lần sau delay
scheduler.schedule(() -> sendReminder(order), 24, TimeUnit.HOURS);

// Chạy định kỳ với fixed rate (bắt đầu tính từ lần trước bắt đầu)
scheduler.scheduleAtFixedRate(
    () -> syncInventory(),
    0,          // initial delay
    5,          // period
    TimeUnit.MINUTES
);

// Chạy định kỳ với fixed delay (bắt đầu tính từ lần trước kết thúc)
// An toàn hơn khi task mất thời gian không đoán được
scheduler.scheduleWithFixedDelay(
    () -> cleanExpiredSessions(),
    0,
    10,
    TimeUnit.MINUTES
);

// Sequential: mỗi bước phải chờ bước trước - chậm
ProductData product = productService.fetch(productId);     // 100ms
StockInfo stock     = inventoryService.check(productId);   // 80ms
PriceData price     = pricingService.calculate(productId); // 60ms
// Tổng: 240ms

// Parallel với CompletableFuture: chạy cùng lúc
CompletableFuture<ProductData> productFuture =
    CompletableFuture.supplyAsync(() -> productService.fetch(productId), executor);

CompletableFuture<StockInfo> stockFuture =
    CompletableFuture.supplyAsync(() -> inventoryService.check(productId), executor);

CompletableFuture<PriceData> priceFuture =
    CompletableFuture.supplyAsync(() -> pricingService.calculate(productId), executor);

// Chờ tất cả hoàn thành
CompletableFuture.allOf(productFuture, stockFuture, priceFuture).join();
// Tổng: ~100ms (bằng task chậm nhất)

ProductResponse response = new ProductResponse(
    productFuture.join(),
    stockFuture.join(),
    priceFuture.join()
);

CompletableFuture<OrderConfirmation> pipeline =
    CompletableFuture.supplyAsync(() -> validateOrder(request), executor)
        .thenApplyAsync(valid -> enrichWithProductData(valid), executor)
        .thenApplyAsync(enriched -> calculateShipping(enriched), executor)
        .thenComposeAsync(order -> paymentService.charge(order), executor) // trả về CF
        .thenApplyAsync(paid -> orderRepository.save(paid), executor)
        .exceptionally(ex -> {
            log.error("Order pipeline failed", ex);
            return OrderConfirmation.failed(ex.getMessage());
        });

// WRONG: dùng common pool
CompletableFuture.supplyAsync(() -> blockingDbCall());

// CORRECT: dùng executor riêng
CompletableFuture.supplyAsync(() -> blockingDbCall(), dbExecutor);

public class SumTask extends RecursiveTask<Long> {
    private static final int THRESHOLD = 10_000; // chia nhỏ đến khi đủ nhỏ
    private final long[] data;
    private final int start, end;

    public SumTask(long[] data, int start, int end) {
        this.data = data; this.start = start; this.end = end;
    }

    @Override
    protected Long compute() {
        if (end - start <= THRESHOLD) {
            // Base case: đủ nhỏ, tính trực tiếp
            long sum = 0;
            for (int i = start; i < end; i++) sum += data[i];
            return sum;
        }
        // Recursive case: chia đôi
        int mid = (start + end) / 2;
        SumTask left  = new SumTask(data, start, mid);
        SumTask right = new SumTask(data, mid, end);

        left.fork();                    // đẩy left vào queue của thread hiện tại
        long rightResult = right.compute(); // chạy right trực tiếp trên thread này
        long leftResult  = left.join(); // lấy kết quả left (hoặc chờ nếu chưa xong)

        return leftResult + rightResult;
    }
}

// Dùng
ForkJoinPool pool = new ForkJoinPool(); // mặc định: số core CPU
long total = pool.invoke(new SumTask(data, 0, data.length));

// parallelStream() dùng ForkJoinPool.commonPool()
List<OrderSummary> summaries = orders.parallelStream()
    .filter(o -> o.getStatus() == COMPLETED)
    .map(orderMapper::toSummary)
    .collect(Collectors.toList());

// VẤN ĐỀ: commonPool được chia sẻ toàn JVM
// Nếu một task trong parallelStream() block (DB call, HTTP call),
// nó chiếm worker của commonPool, ảnh hưởng CompletableFuture và parallelStream khác

// Production fix: dùng ForkJoinPool riêng để cô lập
ForkJoinPool customPool = new ForkJoinPool(4);
List<OrderSummary> summaries = customPool.submit(() ->
    orders.parallelStream()
        .filter(o -> o.getStatus() == COMPLETED)
        .map(orderMapper::toSummary)
        .collect(Collectors.toList())
).get();

// Tạo virtual thread
Thread vt = Thread.ofVirtual().start(() -> processRequest(request));

// Với ExecutorService
try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
    // Tạo virtual thread cho mỗi task - không tốn kém như OS thread
    for (Order order : orders) {
        executor.submit(() -> processOrder(order)); // hàng nghìn task, ổn
    }
} // tự động shutdown khi try-with-resources kết thúc

// Với Platform Thread: cần thread pool sized để tránh thread exhaustion
ExecutorService platformExecutor = Executors.newFixedThreadPool(200);
// Với 200 threads, tối đa 200 concurrent request

// Với Virtual Thread: thread-per-request model trở lại viable
ExecutorService virtualExecutor = Executors.newVirtualThreadPerTaskExecutor();
// Hàng nghìn concurrent request, mỗi cái có virtual thread riêng
// Khi block I/O → unmount, carrier thread xử lý request khác

// WRONG: synchronized block pin virtual thread vào carrier thread
// Khi virtual thread trong synchronized block bị block I/O,
// nó không thể unmount - carrier thread cũng bị block
synchronized (lock) {
    dbConnection.query(sql); // I/O block → pin carrier thread!
}

// CORRECT: dùng ReentrantLock thay vì synchronized
ReentrantLock lock = new ReentrantLock();
lock.lock();
try {
    dbConnection.query(sql); // virtual thread unmount khi block I/O
} finally {
    lock.unlock();
}

// Bug: tạo thread không giới hạn cho mỗi incoming order event
@KafkaListener(topics = "order-events")
public void handleOrder(OrderEvent event) {
    new Thread(() -> {
        enrichWithExternalData(event); // HTTP call 500ms
        orderRepository.save(event);
    }).start(); // tạo thread mới mỗi event, không bounded
}

// Shared counter không được bảo vệ
public class OrderCounter {
    private int count = 0;

    public void increment() {
        count++; // KHÔNG phải atomic! Đây là 3 bước: read, add, write
    }

    public int get() { return count; }
}

Thread 1: đọc count = 100
Thread 2: đọc count = 100   ← đọc trước khi Thread 1 ghi xong
Thread 1: ghi count = 101
Thread 2: ghi count = 101   ← ghi đè kết quả của Thread 1

@Service
public class FlashSaleService {
    private int orderCount = 0; // race condition!
    private static final int MAX_ORDERS = 1000;

    public boolean tryPlaceOrder(Order order) {
        if (orderCount >= MAX_ORDERS) {
            return false; // sold out
        }
        orderCount++; // race condition ở đây
        orderRepository.save(order);
        return true;
    }
}

private final AtomicInteger orderCount = new AtomicInteger(0);

public boolean tryPlaceOrder(Order order) {
    // compareAndSet là atomic: check và increment trong một operation
    int current;
    do {
        current = orderCount.get();
        if (current >= MAX_ORDERS) return false;
    } while (!orderCount.compareAndSet(current, current + 1));

    // Hoặc đơn giản hơn với getAndIncrement + kiểm tra sau
    int slot = orderCount.getAndIncrement();
    if (slot >= MAX_ORDERS) {
        orderCount.decrementAndGet(); // trả lại slot
        return false;
    }
    orderRepository.save(order);
    return true;
}

// Chuyển tiền giữa hai tài khoản - cần atomic!
public class BankAccount {
    private int balance;

    public void transfer(BankAccount target, int amount) {
        this.balance -= amount;
        // Nếu thread khác đọc balance ở đây: thấy tiền đã bị trừ
        // nhưng chưa được cộng vào target. Tổng tiền trong hệ thống bị âm.
        target.balance += amount;
    }
}

public class WalletService {
    private final Map<String, Long> balances = new HashMap<>();

    // Synchronized trên method: lock là 'this' (instance của WalletService)
    public synchronized boolean transfer(String fromId, String toId, long amount) {
        long fromBalance = balances.getOrDefault(fromId, 0L);
        if (fromBalance < amount) return false;

        balances.put(fromId, fromBalance - amount);
        balances.put(toId, balances.getOrDefault(toId, 0L) + amount);
        return true;
    }

    // Đây là vấn đề: lock toàn bộ method → chỉ một transfer được xử lý tại một lúc
    // Ngay cả transfer giữa hai cặp tài khoản hoàn toàn khác nhau cũng phải chờ nhau
}

public boolean transfer(String fromId, String toId, long amount) {
    // Lock chỉ khi thực sự cần thay đổi state
    synchronized (balances) {
        long fromBalance = balances.getOrDefault(fromId, 0L);
        if (fromBalance < amount) return false;
        balances.put(fromId, fromBalance - amount);
        balances.put(toId, balances.getOrDefault(toId, 0L) + amount);
        return true;
    }
    // Các operation không cần lock (logging, audit trail) có thể ở ngoài
}

public class DataProcessor {
    private boolean running = true; // không có volatile

    public void process() {
        while (running) {   // Thread A đọc running từ CPU cache
            doWork();
        }
    }

    public void stop() {
        running = false;    // Thread B ghi vào RAM, nhưng Thread A
                            // vẫn thấy running = true từ cache của nó
                            // → vòng lặp không bao giờ dừng
    }
}

private volatile boolean running = true;

@Component
public class EventProcessor implements DisposableBean {
    private volatile boolean shutdownRequested = false;
    private final ExecutorService executor = Executors.newSingleThreadExecutor();

    @PostConstruct
    public void start() {
        executor.submit(() -> {
            while (!shutdownRequested) {
                try {
                    Event event = eventQueue.poll(100, TimeUnit.MILLISECONDS);
                    if (event != null) processEvent(event);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
            log.info("EventProcessor stopped cleanly");
        });
    }

    @Override
    public void destroy() {
        shutdownRequested = true; // visible đến thread xử lý ngay lập tức
        executor.shutdown();
    }
}

public class DeadlockExample {

    public void transfer(Account from, Account to, long amount) {
        synchronized (from) {              // Thread 1: lock account A
            synchronized (to) {            // Thread 1: chờ lock account B
                from.debit(amount);
                to.credit(amount);
            }
        }
    }
}

// Thread 1: transfer(accountA, accountB, 100)  → lock A, chờ B
// Thread 2: transfer(accountB, accountA, 200)  → lock B, chờ A
// Cả hai chờ nhau mãi mãi

public void transfer(Account from, Account to, long amount) {
    // Dùng System.identityHashCode để xác định thứ tự lock nhất quán
    Account first  = System.identityHashCode(from) < System.identityHashCode(to) ? from : to;
    Account second = first == from ? to : from;

    synchronized (first) {
        synchronized (second) {
            from.debit(amount);
            to.credit(amount);
        }
    }
}

public class PaymentService {
    private final ReentrantLock lock = new ReentrantLock();

    public boolean processPayment(PaymentRequest request) throws InterruptedException {
        // Thử lấy lock, tối đa 2 giây
        if (!lock.tryLock(2, TimeUnit.SECONDS)) {
            // Không lấy được lock sau 2 giây → từ chối thay vì chờ mãi
            log.warn("Payment processing timeout for order {}", request.getOrderId());
            throw new PaymentTimeoutException("System busy, please retry");
        }

        try {
            return doProcessPayment(request);
        } finally {
            lock.unlock(); // LUÔN unlock trong finally
        }
    }
}

ThreadPoolExecutor executor = new ThreadPoolExecutor(
    10,                          // corePoolSize: số thread luôn sống
    50,                          // maximumPoolSize: giới hạn tuyệt đối
    60L, TimeUnit.SECONDS,       // keepAliveTime: thread dư sống bao lâu
    new LinkedBlockingQueue<>(500), // workQueue: giới hạn 500 task đang chờ
    new ThreadPoolExecutor.CallerRunsPolicy() // rejection: caller tự xử lý
);

// Executors.newFixedThreadPool source code:
return new ThreadPoolExecutor(n, n, 0L, TimeUnit.MILLISECONDS,
    new LinkedBlockingQueue<Runnable>()); // KHÔNG CÓ BOUND!

@Bean
public ThreadPoolExecutor orderProcessingExecutor() {
    int cpuCores = Runtime.getRuntime().availableProcessors();

    return new ThreadPoolExecutor(
        cpuCores * 2,          // core: I/O-bound tasks thường dùng 2x cores
        cpuCores * 4,          // max: burst capacity
        30L, TimeUnit.SECONDS,
        new LinkedBlockingQueue<>(1000), // bounded: tránh OOM
        new ThreadFactory() {
            private final AtomicInteger count = new AtomicInteger();
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r, "order-worker-" + count.incrementAndGet());
                t.setDaemon(false); // non-daemon: JVM không thoát khi còn task
                return t;
            }
        },
        (task, executor) -> {
            // Rejection: log + throw để caller biết và có thể retry
            log.error("Order processing queue full, rejecting task");
            throw new RejectedExecutionException("Order processing at capacity");
        }
    );
}

// RequestContext.java
public class RequestContext {
    private static final ThreadLocal<RequestContext> CONTEXT =
        ThreadLocal.withInitial(RequestContext::new);

    private String requestId;
    private String userId;
    private long startTime;

    public static RequestContext current() { return CONTEXT.get(); }

    public static void clear() { CONTEXT.remove(); } // QUAN TRỌNG
}

// OrderFilter.java (Servlet Filter)
public class RequestContextFilter implements Filter {
    @Override
    public void doFilter(ServletRequest req, ServletResponse resp, FilterChain chain)
            throws IOException, ServletException {
        try {
            RequestContext ctx = RequestContext.current();
            ctx.setRequestId(UUID.randomUUID().toString());
            ctx.setUserId(extractUserId((HttpServletRequest) req));
            ctx.setStartTime(System.currentTimeMillis());

            chain.doFilter(req, resp);
        } finally {
            RequestContext.clear(); // BẮT BUỘC để tránh memory leak trong thread pool
        }
    }
}

// Dùng ở bất kỳ đâu trong call stack mà không cần pass parameter
@Service
public class AuditService {
    public void logAction(String action) {
        String requestId = RequestContext.current().getRequestId(); // không cần inject
        log.info("[{}] Action: {}", requestId, action);
    }
}

// BUG: không clear ThreadLocal trong thread pool
ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 10, ...);

executor.submit(() -> {
    RequestContext.current().setUserId("user-42");
    processRequest();
    // Quên gọi RequestContext.clear()!
    // Thread trở về pool với userId = "user-42" còn trong ThreadLocal
    // Request tiếp theo trên thread này đọc được userId của request trước
});

@Service
public class OrderService {

    @Async
    public CompletableFuture<Void> sendConfirmationAsync(Order order) {
        // SecurityContextHolder.getContext().getAuthentication() → null
        // MDC.get("requestId") → null
        // RequestContextHolder.getRequestAttributes() → null
        String userId = getCurrentUserId(); // NullPointerException
        auditService.log(userId, "ORDER_CONFIRMED", order.getId());
        return CompletableFuture.completedFuture(null);
    }
}

public class ContextCopyingTaskDecorator implements TaskDecorator {

    @Override
    public Runnable decorate(Runnable task) {
        // Chạy trên calling thread: capture context hiện tại
        SecurityContext securityCtx = SecurityContextHolder.getContext();
        Map<String, String> mdcCtx   = MDC.getCopyOfContextMap();

        return () -> {
            // Chạy trên worker thread: restore context
            SecurityContextHolder.setContext(securityCtx);
            if (mdcCtx != null) MDC.setContextMap(mdcCtx);
            try {
                task.run();
            } finally {
                SecurityContextHolder.clearContext(); // bắt buộc: thread về pool
                MDC.clear();
            }
        };
    }
}

@Configuration
@EnableAsync
public class AsyncConfig implements AsyncConfigurer {

    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(50);
        executor.setQueueCapacity(500);
        executor.setThreadNamePrefix("async-worker-");
        executor.setTaskDecorator(new ContextCopyingTaskDecorator());
        executor.initialize();
        return executor;
    }
}

SecurityContextHolder.setStrategyName(SecurityContextHolder.MODE_INHERITABLETHREADLOCAL);

@Service
public class OrderProcessingService {

    @Async("orderExecutor")
    public CompletableFuture<OrderResult> processAsync(Order order) {
        // SecurityContext và MDC đã được propagate qua TaskDecorator
        String requestId = MDC.get("requestId"); // có giá trị từ request gốc
        String userId    = SecurityContextHolder.getContext()
                               .getAuthentication().getName();

        log.info("[{}] Processing order {} for user {}",
            requestId, order.getId(), userId);

        OrderResult result = doHeavyProcessing(order);

        auditRepository.save(new AuditEntry(userId, "ORDER_PROCESSED",
            order.getId(), requestId));

        return CompletableFuture.completedFuture(result);
    }
}

// WRONG trong Quarkus: block event loop
@GET
@Path("/orders/{id}")
public Order getOrder(String id) {
    return orderRepository.findById(id).await().indefinitely(); // blocking!
    // event loop bị chặn → toàn bộ server không nhận request khác
}

// ĐÚNG: annotate blocking method
@GET
@Path("/orders/{id}")
@Blocking  // Quarkus dispatch sang worker thread tự động
public Order getOrder(String id) {
    return orderRepository.findById(id); // JDBC blocking, OK trên worker thread
}

@ApplicationScoped
public class BackgroundJobService {

    @Inject
    ManagedExecutorService executor; // inject, không tự tạo

    @Inject
    SecurityIdentity identity;

    public void scheduleInventoryUpdate(List<Product> products) {
        executor.submit(() -> {
            // CDI context có sẵn - không cần setup thủ công
            String currentUser = identity.getPrincipal().getName(); // hoạt động
            products.forEach(inventoryService::update); // @Transactional works
        });
    }
}

@ApplicationScoped
public class ReportService {

    @Inject
    ManagedExecutorService executor;

    @Inject
    InjectableContext requestContext; // Quarkus CDI request context

    public void generateReportAsync(ReportRequest request) {
        executor.submit(() -> {
            // Request scope mặc định không active trong background thread
            // Activate thủ công:
            requestContext.activate();
            try {
                doGenerateReport(request); // code dùng @RequestScoped beans
            } finally {
                requestContext.terminate(); // cleanup
            }
        });
    }
}

@GET
@Path("/orders/process")
public Uni<OrderResult> processOrder(@Valid OrderRequest request) {
    return Uni.createFrom().item(request)
        .onItem().transformToUni(req ->
            // Vẫn trên event loop: validate, nhẹ
            Uni.createFrom().item(validateAndEnrich(req))
        )
        .emitOn(executor)                    // switch sang worker thread
        .onItem().transform(enriched -> {
            // Trên worker thread: blocking DB call
            // SecurityIdentity, CDI context được propagate bởi SmallRye
            return orderRepository.save(enriched);
        })
        .emitOn(Infrastructure.getDefaultWorkerPool())
        .onItem().transform(saved -> {
            notificationService.sendConfirmation(saved);
            return new OrderResult(saved.getId());
        });
}