Everything About Threads in Java

// Single-threaded server - sequential processing
while (true) {
    Request req = acceptConnection();  // receive request
    processOrder(req);                 // 250ms+ per request
    // Next request must WAIT until the current one finishes
}

// Option 1: Subclass Thread
// Problem: cannot subclass another class, harder to test, tightly coupled
class OrderProcessor extends Thread {
    @Override
    public void run() { processOrder(); }
}
new OrderProcessor().start();

// Option 2: Runnable - separates the task from the execution mechanism
Runnable task = () -> processOrder();
new Thread(task).start();

ExecutorService executor = Executors.newFixedThreadPool(10);

// Callable: returns a result and can throw Exception
Callable<ProductData> task = () -> {
    return externalApiClient.fetchProduct(productId); // may throw IOException
};

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

// Do other work while the task runs asynchronously
doOtherWork();

// Get the result, blocking if not yet done
try {
    ProductData data = future.get(5, TimeUnit.SECONDS); // timeout prevents blocking forever
} catch (TimeoutException e) {
    future.cancel(true); // interrupt the task on timeout
    throw new ServiceUnavailableException("External API timeout");
} catch (ExecutionException e) {
    throw new RuntimeException("Task failed", e.getCause());
}

// Bug: get() without timeout
ProductData data = future.get(); // blocks forever if the external API hangs
// The user's HTTP request also blocks forever -> load balancer timeout
// Thread held in pool -> pool exhaustion -> entire service hangs

ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(4);

// Run once after a delay
scheduler.schedule(() -> sendReminder(order), 24, TimeUnit.HOURS);

// Run periodically with a fixed rate (period starts from when the previous run started)
scheduler.scheduleAtFixedRate(
    () -> syncInventory(),
    0,          // initial delay
    5,          // period
    TimeUnit.MINUTES
);

// Run periodically with a fixed delay (period starts from when the previous run ended)
// Safer when task duration is unpredictable
scheduler.scheduleWithFixedDelay(
    () -> cleanExpiredSessions(),
    0,
    10,
    TimeUnit.MINUTES
);

// Sequential: each step waits for the previous one - slow
ProductData product = productService.fetch(productId);     // 100ms
StockInfo stock     = inventoryService.check(productId);   // 80ms
PriceData price     = pricingService.calculate(productId); // 60ms
// Total: 240ms

// Parallel with CompletableFuture: run concurrently
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);

// Wait for all to complete
CompletableFuture.allOf(productFuture, stockFuture, priceFuture).join();
// Total: ~100ms (limited by the slowest task)

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) // returns CF
        .thenApplyAsync(paid -> orderRepository.save(paid), executor)
        .exceptionally(ex -> {
            log.error("Order pipeline failed", ex);
            return OrderConfirmation.failed(ex.getMessage());
        });

// WRONG: uses common pool
CompletableFuture.supplyAsync(() -> blockingDbCall());

// CORRECT: uses a dedicated executor
CompletableFuture.supplyAsync(() -> blockingDbCall(), dbExecutor);

public class SumTask extends RecursiveTask<Long> {
    private static final int THRESHOLD = 10_000; // split until small enough
    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: small enough, compute directly
            long sum = 0;
            for (int i = start; i < end; i++) sum += data[i];
            return sum;
        }
        // Recursive case: split in half
        int mid = (start + end) / 2;
        SumTask left  = new SumTask(data, start, mid);
        SumTask right = new SumTask(data, mid, end);

        left.fork();                    // push left onto current thread's queue
        long rightResult = right.compute(); // run right directly on this thread
        long leftResult  = left.join(); // get left result (wait if not done)

        return leftResult + rightResult;
    }
}

// Usage
ForkJoinPool pool = new ForkJoinPool(); // default: number of CPU cores
long total = pool.invoke(new SumTask(data, 0, data.length));

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

// PROBLEM: commonPool is shared across the entire JVM
// If a task in parallelStream() blocks (DB call, HTTP call),
// it occupies a commonPool worker and affects all CompletableFutures and other parallelStreams

// Production fix: use a dedicated ForkJoinPool for isolation
ForkJoinPool customPool = new ForkJoinPool(4);
List<OrderSummary> summaries = customPool.submit(() ->
    orders.parallelStream()
        .filter(o -> o.getStatus() == COMPLETED)
        .map(orderMapper::toSummary)
        .collect(Collectors.toList())
).get();

// Create a virtual thread
Thread vt = Thread.ofVirtual().start(() -> processRequest(request));

// With ExecutorService
try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
    // Creates a virtual thread for each task - not as expensive as OS threads
    for (Order order : orders) {
        executor.submit(() -> processOrder(order)); // thousands of tasks, no problem
    }
} // automatically shuts down when try-with-resources exits

// With platform threads: pool must be sized to avoid thread exhaustion
ExecutorService platformExecutor = Executors.newFixedThreadPool(200);
// With 200 threads, at most 200 concurrent requests

// With virtual threads: thread-per-request model becomes viable again
ExecutorService virtualExecutor = Executors.newVirtualThreadPerTaskExecutor();
// Thousands of concurrent requests, each with its own virtual thread
// When blocked on I/O -> unmounted, carrier thread processes another request

// WRONG: synchronized block pins the virtual thread to the carrier thread
// When a virtual thread inside a synchronized block is blocked on I/O,
// it cannot unmount - the carrier thread is also blocked
synchronized (lock) {
    dbConnection.query(sql); // I/O block -> pins carrier thread!
}

// CORRECT: use ReentrantLock instead of synchronized
ReentrantLock lock = new ReentrantLock();
lock.lock();
try {
    dbConnection.query(sql); // virtual thread unmounts when blocked on I/O
} finally {
    lock.unlock();
}

// Bug: creates an unbounded number of threads for each incoming order event
@KafkaListener(topics = "order-events")
public void handleOrder(OrderEvent event) {
    new Thread(() -> {
        enrichWithExternalData(event); // HTTP call 500ms
        orderRepository.save(event);
    }).start(); // creates a new thread per event, unbounded
}

// Unprotected shared counter
public class OrderCounter {
    private int count = 0;

    public void increment() {
        count++; // NOT atomic! This is 3 steps: read, add, write
    }

    public int get() { return count; }
}

Thread 1: reads count = 100
Thread 2: reads count = 100   <- reads before Thread 1 finishes writing
Thread 1: writes count = 101
Thread 2: writes count = 101  <- overwrites Thread 1's result

@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 here
        orderRepository.save(order);
        return true;
    }
}

private final AtomicInteger orderCount = new AtomicInteger(0);

public boolean tryPlaceOrder(Order order) {
    // compareAndSet is atomic: check and increment in a single operation
    int current;
    do {
        current = orderCount.get();
        if (current >= MAX_ORDERS) return false;
    } while (!orderCount.compareAndSet(current, current + 1));

    // Or more simply with getAndIncrement + check after
    int slot = orderCount.getAndIncrement();
    if (slot >= MAX_ORDERS) {
        orderCount.decrementAndGet(); // return the slot
        return false;
    }
    orderRepository.save(order);
    return true;
}

// Transfer money between two accounts - must be atomic!
public class BankAccount {
    private int balance;

    public void transfer(BankAccount target, int amount) {
        this.balance -= amount;
        // If another thread reads balance here: it sees the money already deducted
        // but not yet added to target. The total money in the system appears negative.
        target.balance += amount;
    }
}

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

    // synchronized on method: lock is 'this' (the WalletService instance)
    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;
    }

    // The problem: locking the whole method means only one transfer runs at a time
    // Even transfers between completely different pairs of accounts must queue up
}

public boolean transfer(String fromId, String toId, long amount) {
    // Lock only when actually modifying 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;
    }
    // Operations that do not need locking (logging, audit trail) can go outside
}

public class DataProcessor {
    private boolean running = true; // no volatile

    public void process() {
        while (running) {   // Thread A reads 'running' from CPU cache
            doWork();
        }
    }

    public void stop() {
        running = false;    // Thread B writes to RAM, but Thread A
                            // still sees running = true from its cache
                            // -> the loop never stops
    }
}

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 to the processing thread immediately
        executor.shutdown();
    }
}

public class DeadlockExample {

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

// Thread 1: transfer(accountA, accountB, 100)  -> locks A, waits for B
// Thread 2: transfer(accountB, accountA, 200)  -> locks B, waits for A
// Both wait for each other forever

public void transfer(Account from, Account to, long amount) {
    // Use System.identityHashCode to determine a consistent lock order
    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 {
        // Try to acquire the lock for up to 2 seconds
        if (!lock.tryLock(2, TimeUnit.SECONDS)) {
            // Could not acquire lock after 2 seconds -> reject instead of waiting forever
            log.warn("Payment processing timeout for order {}", request.getOrderId());
            throw new PaymentTimeoutException("System busy, please retry");
        }

        try {
            return doProcessPayment(request);
        } finally {
            lock.unlock(); // ALWAYS unlock in finally
        }
    }
}

ThreadPoolExecutor executor = new ThreadPoolExecutor(
    10,                          // corePoolSize: threads always alive
    50,                          // maximumPoolSize: absolute upper limit
    60L, TimeUnit.SECONDS,       // keepAliveTime: how long extra threads stay alive
    new LinkedBlockingQueue<>(500), // workQueue: limit of 500 pending tasks
    new ThreadPoolExecutor.CallerRunsPolicy() // rejection: caller handles the task itself
);

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

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

    return new ThreadPoolExecutor(
        cpuCores * 2,          // core: I/O-bound tasks typically benefit from 2x cores
        cpuCores * 4,          // max: burst capacity
        30L, TimeUnit.SECONDS,
        new LinkedBlockingQueue<>(1000), // bounded: prevents 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 does not exit while tasks remain
                return t;
            }
        },
        (task, executor) -> {
            // Rejection: log and throw so the caller knows and can 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(); } // IMPORTANT
}

// 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(); // REQUIRED to prevent memory leaks in thread pools
        }
    }
}

// Usable anywhere in the call stack without passing parameters
@Service
public class AuditService {
    public void logAction(String action) {
        String requestId = RequestContext.current().getRequestId(); // no injection needed
        log.info("[{}] Action: {}", requestId, action);
    }
}

// BUG: not clearing ThreadLocal in a thread pool
ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 10, ...);

executor.submit(() -> {
    RequestContext.current().setUserId("user-42");
    processRequest();
    // Forgot to call RequestContext.clear()!
    // Thread returns to pool with userId = "user-42" still in its ThreadLocal
    // The next request on this thread reads the previous request's userId
});

@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) {
        // Runs on the calling thread: capture current context
        SecurityContext securityCtx = SecurityContextHolder.getContext();
        Map<String, String> mdcCtx   = MDC.getCopyOfContextMap();

        return () -> {
            // Runs on the worker thread: restore context
            SecurityContextHolder.setContext(securityCtx);
            if (mdcCtx != null) MDC.setContextMap(mdcCtx);
            try {
                task.run();
            } finally {
                SecurityContextHolder.clearContext(); // required: thread returns to 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 and MDC have been propagated via TaskDecorator
        String requestId = MDC.get("requestId"); // has value from the original request
        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 in Quarkus: blocking the event loop
@GET
@Path("/orders/{id}")
public Order getOrder(String id) {
    return orderRepository.findById(id).await().indefinitely(); // blocking!
    // event loop is blocked -> the entire server stops accepting requests
}

// CORRECT: annotate blocking methods
@GET
@Path("/orders/{id}")
@Blocking  // Quarkus automatically dispatches to a worker thread
public Order getOrder(String id) {
    return orderRepository.findById(id); // JDBC blocking, fine on a worker thread
}

@ApplicationScoped
public class BackgroundJobService {

    @Inject
    ManagedExecutorService executor; // injected, not manually created

    @Inject
    SecurityIdentity identity;

    public void scheduleInventoryUpdate(List<Product> products) {
        executor.submit(() -> {
            // CDI context is available - no manual setup needed
            String currentUser = identity.getPrincipal().getName(); // works
            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 is not active in background threads by default
            // Activate it manually:
            requestContext.activate();
            try {
                doGenerateReport(request); // code uses @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 ->
            // Still on event loop: validate, lightweight
            Uni.createFrom().item(validateAndEnrich(req))
        )
        .emitOn(executor)                    // switch to worker thread
        .onItem().transform(enriched -> {
            // On worker thread: blocking DB call
            // SecurityIdentity, CDI context propagated by SmallRye
            return orderRepository.save(enriched);
        })
        .emitOn(Infrastructure.getDefaultWorkerPool())
        .onItem().transform(saved -> {
            notificationService.sendConfirmation(saved);
            return new OrderResult(saved.getId());
        });
}