How do I create custom thread factories?

By I Wayan Saryada in Concurrency, Util Package 0 Comment

Creating a custom thread factory in Java is a powerful way to manage how threads are initialized. Instead of using the default factory, you can customize thread names (vital for debugging!), set priority levels, or even create daemon threads.

To do this, you need to implement the java.util.concurrent.ThreadFactory interface.

1. Implement the ThreadFactory Interface

The interface has a single method: newThread(Runnable r). Here is a clean, reusable example:

package org.kodejava.util.concurrent;

import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

public class CustomThreadFactory implements ThreadFactory {
    private final String namePrefix;
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private final boolean daemon;

    public CustomThreadFactory(String poolName, boolean daemon) {
        this.namePrefix = poolName + "-worker-";
        this.daemon = daemon;
    }

    @Override
    public Thread newThread(Runnable r) {
        Thread t = new Thread(r, namePrefix + threadNumber.getAndIncrement());
        t.setDaemon(daemon);
        t.setPriority(Thread.NORM_PRIORITY);
        return t;
    }
}

2. Use the Factory with an ExecutorService

Once you’ve defined your factory, you can pass it to any ThreadPoolExecutor or static Executors factory method. This ensures every thread created by that pool follows your rules.

package org.kodejava.util.concurrent;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ThreadFactoryDemo {
    public static void main(String[] args) {
        // Create the factory
        CustomThreadFactory factory = new CustomThreadFactory("OrderProcessor", false);

        // Pass it to a Fixed Thread Pool
        ExecutorService executor = Executors.newFixedThreadPool(3, factory);

        for (int i = 0; i < 5; i++) {
            executor.submit(() -> {
                System.out.println("Running task in: " + Thread.currentThread().getName());
            });
        }

        executor.shutdown();
    }
}

Why use a custom factory?

  • Identifiability: In thread dumps or logs, seeing OrderProcessor-worker-1 is much more helpful than pool-1-thread-1.
  • Security & Cleanup: You can set setDaemon(true) for background cleanup tasks so they don’t prevent the JVM from shutting down.
  • Context: You can use the factory to inject ThreadLocal variables or set a custom UncaughtExceptionHandler for all threads in a pool.

How do I use ExecutorService.invokeAll?

By I Wayan Saryada in Concurrency, Util Package 0 Comment

Hello! ExecutorService.invokeAll is a powerful method when you have a collection of tasks and need to wait until every single one of them finishes before moving forward.

Here’s a breakdown of how it works and how to use it effectively.

What does invokeAll do?

  1. Executes a collection of tasks: It takes a Collection of Callable<T> objects.
  2. Blocks until completion: Unlike submit(), which returns immediately, invokeAll is blocking. It will not return until all tasks in the collection have completed (either normally or by throwing an exception).
  3. Returns a list of Futures: It returns a List<Future<T>> that holds the results (or exceptions) of the tasks, in the same order they were provided in the input collection.

Basic Usage Pattern

Here is a clean example of how to implement it:

package org.kodejava.util.concurrent;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.*;

public class InvokeAllDemo {
    public static void main(String[] args) {
        // 1. Create your ExecutorService
        try (ExecutorService executor = Executors.newFixedThreadPool(3)) {

            // 2. Define your tasks (Callable returns a value)
            List<Callable<String>> tasks = Arrays.asList(
                    () -> { Thread.sleep(500); return "Result A"; },
                    () -> { Thread.sleep(1000); return "Result B"; },
                    () -> { Thread.sleep(200); return "Result C"; }
            );

            try {
                // 3. Invoke all tasks. Execution stops here until all are done.
                System.out.println("Executing tasks...");
                List<Future<String>> futures = executor.invokeAll(tasks);

                // 4. Process the results
                for (Future<String> future : futures) {
                    // Future.get() will not block here because invokeAll already waited
                    System.out.println("Task output: " + future.get());
                }
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        } // Executor closes automatically with try-with-resources (Java 19+)
    }
}

Important Considerations

  • Order Preservation: The returned List<Future> maintains the same order as the input task list. futures.get(0) will always correspond to tasks.get(0).
  • Timeouts: There is an overloaded version: invokeAll(tasks, timeout, unit). If the timeout expires, tasks that haven’t finished are canceled, and the method returns the list of futures (some will be marked as canceled).
  • Exceptions: If a task throws an exception, invokeAll doesn’t fail. Instead, that specific Future.get() will throw an ExecutionException.
  • Blocking Behavior: Since invokeAll blocks the calling thread, avoid calling it on a thread that needs to stay responsive (like a UI thread or a primary event loop) without careful planning.

When to use it vs invokeAny?

  • Use invokeAll when you need the results of everything you started.
  • Use invokeAny when you have multiple ways to get a result, and you only care about the first one that finishes successfully (it cancels the rest).

Happy coding! If you’re working within a Spring environment, you might also want to look into @Async for higher-level abstraction, but for raw concurrency control, invokeAll is a classic choice.

How do I use Atomic variables from java.util.concurrent.atomic?

By I Wayan Saryada in Cryptography, Util Package 0 Comment

Atomic variables in java.util.concurrent.atomic are designed for lock-free, thread-safe operations on single variables. They leverage low-level CPU instructions like Compare-And-Swap (CAS) to ensure data consistency without the overhead of synchronized blocks.

Here’s a guide on how to use the most common atomic classes.

1. AtomicInteger and AtomicLong

These are used for numeric counters or IDs. Instead of using ++ (which is not atomic), you use methods like incrementAndGet().

package org.kodejava.util.concurrent;

import java.util.concurrent.atomic.AtomicInteger;

public class CounterExample {
    private final AtomicInteger counter = new AtomicInteger(0);

    public void increment() {
        // Atomically increments by one and returns the new value
        int newValue = counter.incrementAndGet();
        System.out.println("Current Value: " + newValue);
    }

    public int getValue() {
        return counter.get();
    }
}

2. AtomicBoolean

Useful for flags that need to be checked and updated across threads, such as a “running” state.

package org.kodejava.util.concurrent;

import java.util.concurrent.atomic.AtomicBoolean;

public class Worker {
    private final AtomicBoolean initialized = new AtomicBoolean(false);

    public void init() {
        // compareAndSet(expectedValue, newValue)
        // Only sets to true if it was currently false
        if (initialized.compareAndSet(false, true)) {
            System.out.println("Performing one-time initialization...");
        }
    }
}

3. AtomicReference

Used to wrap any object reference. This is great for implementing non-blocking algorithms where you need to update an entire object state atomically.

package org.kodejava.util.concurrent;

import java.util.concurrent.atomic.AtomicReference;

public class StateManager {
    private final AtomicReference<String> status = new AtomicReference<>("IDLE");

    public void updateStatus(String oldStatus, String newStatus) {
        boolean success = status.compareAndSet(oldStatus, newStatus);
        if (success) {
            System.out.println("Status changed to: " + newStatus);
        }
    }
}

4. Advanced Accumulators (LongAdder)

If you have a very high-contention environment (many threads constantly updating a sum), LongAdder is generally faster than AtomicLong because it maintains internal cells to reduce contention.

package org.kodejava.util.concurrent;

import java.util.concurrent.atomic.LongAdder;

public class HighContentionCounter {
    private final LongAdder adder = new LongAdder();

    public void add() {
        adder.increment();
    }

    public long getTotal() {
        return adder.sum();
    }
}

Key Methods to Remember

  • get() / set(): Read or write the value (similar to volatile).
  • lazySet(): Eventually sets the value; faster but doesn’t guarantee immediate visibility to other threads.
  • compareAndSet(expect, update): The heart of atomic variables. Updates only if the current value matches expect.
  • getAndAccumulate(delta, accumulatorFunction): (Java 8+) Allows complex atomic updates using a Lambda.

When to use them?

  • Use them for simple counters, sequence generators, or flags.
  • Avoid them if you need to update multiple dependent variables at once; in that case, a ReentrantLock or synchronized block is safer to ensure the entire operation is atomic.

How do I use ReentrantLock vs synchronized?

By I Wayan Saryada in Concurrency, Util Package 0 Comment

In Java, both synchronized and ReentrantLock are used to manage thread safety, but they offer different levels of control and flexibility.

1. synchronized Keyword

The synchronized keyword is the built-in mechanism in Java for mutual exclusion. It is simpler to use because the JVM automatically handles acquiring and releasing the lock.

  • Usage: Can be applied to methods or used as a block.
  • Automatic Release: The lock is automatically released when the thread exits the block or method, even if an exception occurs.
  • Performance: In modern JVMs, synchronized is highly optimized (biased locking, lightweight locking) and often performs just as well as ReentrantLock.

Example:

package org.kodejava.util.concurrent;

public class Counter {
    private int count = 0;

    public synchronized void increment() {
        count++;
    }

    public void incrementBlock() {
        synchronized (this) {
            count++;
        }
    }
}

2. ReentrantLock Class

ReentrantLock (from java.util.concurrent.locks) is an explicit lock implementation that provides advanced features not available with synchronized.

  • Manual Control: You must explicitly call lock() and unlock(). It is critical to use a try-finally block to ensure the lock is released.
  • Fairness: You can create a “fair” lock that gives preference to the longest-waiting thread.
  • Try Lock: tryLock() allows a thread to attempt to acquire the lock without blocking indefinitely.
  • Interruptible: A thread waiting for a ReentrantLock can be interrupted via lockInterruptibly().

Example:

package org.kodejava.util.concurrent;

import java.util.concurrent.locks.ReentrantLock;

public class LockCounter {
    private final ReentrantLock lock = new ReentrantLock();
    private int count = 0;

    public void increment() {
        lock.lock();
        try {
            count++;
        } finally {
            // Always unlock in finally to prevent deadlocks
            lock.unlock();
        }
    }
}

Key Differences

Feature synchronized ReentrantLock
Ease of Use Very easy; handled by JVM. Requires manual try-finally.
Fairness Not supported. Supported (optional constructor parameter).
Flexibility Rigid block structure. Highly flexible (can lock in one method, unlock in another).
Non-blocking No (thread always waits). Yes, via tryLock().
Interruptibility No. Yes, via lockInterruptibly().

Which should you choose?

  • Use synchronized by default. It’s cleaner, less error-prone, and sufficient for most basic thread-safety needs.
  • Use ReentrantLock when you need advanced features like:
    • Timing out while waiting for a lock (tryLock(timeout)).
    • Allowing the lock attempt to be interrupted.
    • Using a “Fair” lock strategy.
    • Complex locking structures that aren’t strictly nested.

If you are debugging concurrency issues, IntelliJ IDEA provides tools to inspect thread states and detect blocked monitors [1]. You can also use the Threads tab in the Debugger to see which thread owns a monitor or is waiting for one [2].

How do I use CountDownLatch for coordination?

By I Wayan Saryada in Concurrency, Util Package 0 Comment

A CountDownLatch is a synchronization aid that allows one or more threads to wait until a set of operations being performed in other threads completes. It’s initialized with a given count, and the await() methods block until the current count reaches zero due to invocations of the countDown() method.

Here is a practical guide on how to coordinate threads using CountDownLatch.

1. Basic Coordination Pattern

The most common use case is having a main thread wait for several worker threads to finish their tasks.

package org.kodejava.util.concurrent;

import java.util.concurrent.CountDownLatch;

public class LatchCoordination {
    public static void main(String[] args) throws InterruptedException {
        int numberOfWorkers = 3;
        // 1. Initialize with the number of events to wait for
        CountDownLatch latch = new CountDownLatch(numberOfWorkers);

        for (int i = 0; i < numberOfWorkers; i++) {
            new Thread(new Worker(latch, "Worker-" + i)).start();
        }

        System.out.println("Main thread is waiting...");
        // 2. Wait until the count reaches zero
        latch.await();

        System.out.println("All workers finished. Main thread proceeding.");
    }
}

class Worker implements Runnable {
    private final CountDownLatch latch;
    private final String name;

    Worker(CountDownLatch latch, String name) {
        this.latch = latch;
        this.name = name;
    }

    @Override
    public void run() {
        try {
            System.out.println(name + " is performing work...");
            Thread.sleep(1000); // Simulate work
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            // 3. Crucial: Always decrement the count in a finally block
            latch.countDown();
            System.out.println(name + " finished. Remaining: " + latch.getCount());
        }
    }
}

2. Key Methods to Remember

  • new CountDownLatch(int count): Sets the initial count. This count cannot be reset. If you need to reset the count, consider using a CyclicBarrier.
  • countDown(): Decrements the count of the latch, releasing all waiting threads if the count reaches zero.
  • await(): Causes the current thread to wait until the latch has counted down to zero, unless the thread is interrupted.
  • await(long timeout, TimeUnit unit): A safer version that waits for a specific duration. It returns true if the count reached zero and false if the waiting time elapsed before the count reached zero.

3. Advanced Use: The “Starting Gun”

You can also use a CountDownLatch with a count of 1 as a “starting gun” to release many threads at the exact same moment (useful for load testing).

CountDownLatch startSignal = new CountDownLatch(1);

for (int i = 0; i < 10; i++) {
    new Thread(() -> {
        try {
            startSignal.await(); // All threads stall here
            // Do concurrent work...
        } catch (InterruptedException e) { /* handle */ }
    }).start();
}

// All threads start simultaneously when this is called
startSignal.countDown();

Best Practices

  • Always use finally: Place countDown() inside a finally block to ensure the latch is decremented even if a worker thread encounters an exception. Otherwise, the waiting thread might block indefinitely.
  • Don’t reuse: Once a CountDownLatch reaches zero, it cannot be reused. If your algorithm requires repeating the synchronization point, use a CyclicBarrier or Phaser.
  • Check the count: You can use latch.getCount() for debugging or monitoring, but don’t use it to make control flow decisions in production code as it is volatile.