Executors and Concurrent Collections

Introduction

Modern Java applications frequently perform many tasks at the same time: processing web requests, reading files, querying databases, handling messages, and calculating results. Creating a new thread for every task is simple, but it is usually inefficient and difficult to control. Threads consume memory, require scheduling, and can overwhelm a processor or an external service such as a database.

The java.util.concurrent package provides higher-level abstractions for managing concurrent work. ExecutorService separates task submission from thread management. Future represents a result that may become available later. The Fork/Join framework supports recursive, divide-and-conquer computation. Concurrent collections provide safe access to shared data without requiring every operation to be manually protected by a single lock.

Concurrency does not automatically improve performance. A sound design must consider task size, CPU cores, blocking operations, memory visibility, contention, ordering, cancellation, shutdown, and error handling. This chapter develops these ideas and applies them to a producer-consumer system.

Learning Objectives

After studying this chapter, you should be able to:

• explain the purpose and lifecycle of an ExecutorService;
• select an appropriate thread-pool type;
• submit tasks and manage results with Future;
• distinguish CPU-bound, blocking, and scheduled workloads;
• use Fork/Join and RecursiveTask for divide-and-conquer problems;
• select suitable concurrent collections;
• implement producer-consumer designs with BlockingQueue;
• use ConcurrentHashMap and copy-on-write collections correctly;
• identify race conditions, visibility problems, and unsafe compound actions;
• design systems with cancellation, shutdown, bounded resources, and failure handling.

1. From Threads to Executors

A low-level thread-based design couples three responsibilities: defining work, creating threads, and deciding when threads terminate. Executors separate these responsibilities. A task is represented by Runnable when it produces no result, or by Callable<V> when it returns a value of type V and may throw a checked exception.

Runnable auditTask = () -> System.out.println("Audit completed");
Callable<Integer> countTask = () -> 42;

An executor accepts tasks and arranges for them to run. A pool normally contains worker threads that repeatedly take tasks from an internal queue. Reusing workers avoids the cost of creating a new thread for every operation and permits the application to limit concurrency.

Executor interfaces

Executor defines one method, execute(Runnable). ExecutorService extends it with task submission, result retrieval, shutdown, and bulk operations. ScheduledExecutorService adds delayed and periodic execution.

Common factory methods include:

In production code, explicitly constructing ThreadPoolExecutor is often preferable because it permits a bounded queue, a rejection policy, and a clear resource limit.

Pool sizing

For CPU-bound tasks, a pool near the number of available processor cores is a reasonable starting point. For blocking tasks, more workers may be useful because some workers are waiting, but the number must be limited by external resources such as database connections or service quotas. Pool size is therefore an empirical design parameter, not a universal constant.

A useful conceptual approximation for a blocking workload is:

Here, N_threads is a starting estimate, N_cores is the number of usable processor cores, W is average waiting time, and C is average active computation time. The approximation assumes reasonably independent tasks and stable workload characteristics. It is not a guarantee: excessive threads can increase context switching, memory consumption, queueing, and contention.

Lifecycle and shutdown

An executor does not normally stop merely because the submitting method has returned. Worker threads may remain alive. A well-managed application shuts the executor down when it is no longer needed.

ExecutorService service = Executors.newFixedThreadPool(4);
try {
    service.submit(() -> processFile("data.csv"));
} finally {
    service.shutdown();
}

shutdown() rejects new tasks but allows submitted tasks to finish. shutdownNow() attempts to interrupt running tasks and returns tasks that never started; interruption is cooperative, so a task must respond to interruption. A robust termination sequence may use awaitTermination and then escalate carefully.

service.shutdown();
try {
    if (!service.awaitTermination(30, TimeUnit.SECONDS)) {
        service.shutdownNow();
    }
} catch (InterruptedException ex) {
    service.shutdownNow();
    Thread.currentThread().interrupt();
}

Restoring the interrupt flag is important because interruption is a communication mechanism between threads.

2. Future and Completion Management

Submitting a Callable<V> returns Future<V>. A future supports status inspection, blocking result retrieval, cancellation, and exception reporting.

Future<String> future = service.submit(() -> loadCustomerName(17));
try {
    String name = future.get(2, TimeUnit.SECONDS);
    System.out.println(name);
} catch (TimeoutException ex) {
    future.cancel(true);
} catch (ExecutionException ex) {
    Throwable cause = ex.getCause();
    cause.printStackTrace();
} catch (InterruptedException ex) {
    Thread.currentThread().interrupt();
}

get() blocks until completion. get(timeout, unit) bounds the wait. If the task throws an exception, get() throws ExecutionException; the original exception is available through getCause(). Cancellation is a request, not a guarantee. cancel(true) may interrupt a running task, but code that ignores interruption may continue.

A common design error is submitting many tasks and immediately calling get() on each future in submission order. If the first task is slow, completed later tasks cannot be processed promptly. ExecutorCompletionService allows results to be retrieved in completion order.

CompletionService<Result> completion =
    new ExecutorCompletionService<>(service);
for (Request request : requests) {
    completion.submit(() -> handle(request));
}
for (int i = 0; i < requests.size(); i++) {
    Result result = completion.take().get();
    store(result);
}

The application should define what happens when one task fails: continue with partial results, cancel related tasks, retry, or fail the whole operation. Retrying non-idempotent operations can create duplicate effects and therefore requires special care.

3. Bounded Executors and Back Pressure

An unbounded task queue can hide overload. Producers continue submitting work while memory usage increases. A bounded ThreadPoolExecutor makes capacity explicit.

int workers = 4;
int capacity = 100;
BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(capacity);
ThreadPoolExecutor executor = new ThreadPoolExecutor(
    workers, workers,
    0L, TimeUnit.MILLISECONDS,
    queue,
    new ThreadPoolExecutor.CallerRunsPolicy());

CallerRunsPolicy makes the submitting thread execute rejected work when the pool and queue are full. This can slow producers and create back pressure. Other policies include rejecting immediately, discarding a task, or discarding the oldest queued task. The correct choice depends on whether work is essential, replaceable, or time-sensitive.

Bounded capacity is a form of governance: it prevents an internal component from consuming unlimited memory or overwhelming a downstream system. Metrics such as queue length, active count, completed tasks, rejection count, and task latency should be monitored.

4. Fork/Join Parallelism

The Fork/Join framework is designed for computations that can be recursively divided into smaller independent subtasks. A RecursiveTask<V> returns a value; a RecursiveAction does not.

class SumTask extends RecursiveTask<Long> {
    private static final int THRESHOLD = 10_000;
    private final int[] values;
    private final int from;
    private final int to;

    SumTask(int[] values, int from, int to) {
        this.values = values;
        this.from = from;
        this.to = to;
    }

    @Override
    protected Long compute() {
        if (to - from <= THRESHOLD) {
            long total = 0;
            for (int i = from; i < to; i++) total += values[i];
            return total;
        }
        int middle = (from + to) / 2;
        SumTask left = new SumTask(values, from, middle);
        SumTask right = new SumTask(values, middle, to);
        left.fork();
        long rightResult = right.compute();
        long leftResult = left.join();
        return leftResult + rightResult;
    }
}

Work stealing allows an idle worker to take tasks from another worker's deque. The pattern fork one subtask, compute the other directly, and then join is often more efficient than forking both subtasks.

The threshold is important. If tasks become too small, scheduling overhead can exceed the computational benefit. Fork/Join is primarily intended for CPU-bound, non-blocking work. Blocking I/O inside the common pool can delay unrelated tasks. A dedicated executor or a carefully designed managed-blocking strategy may be more appropriate.

Parallel decomposition is beneficial only when the work is sufficiently large and independent. Shared mutable state, uneven partitions, synchronization, and memory bandwidth can reduce or eliminate speedup.

5. BlockingQueue and the Producer-Consumer Pattern

A producer creates work and places it in a queue. A consumer removes and processes it. BlockingQueue coordinates these roles safely:

• put waits when the queue is full;
• take waits when the queue is empty;
• offer and poll can return immediately or use a timeout;
• a bounded queue supplies natural back pressure.

BlockingQueue<String> jobs = new ArrayBlockingQueue<>(50);

Runnable producer = () -> {
    try {
        for (String job : readJobs()) jobs.put(job);
        jobs.put("STOP");
    } catch (InterruptedException ex) {
        Thread.currentThread().interrupt();
    }
};

Runnable consumer = () -> {
    try {
        while (true) {
            String job = jobs.take();
            if (job.equals("STOP")) break;
            process(job);
        }
    } catch (InterruptedException ex) {
        Thread.currentThread().interrupt();
    }
};

With multiple consumers, one sentinel per consumer is generally required, or a separate coordinated shutdown mechanism must be used. A sentinel must not be confused with valid business data. In many systems, cancellation plus executor shutdown is safer than relying only on sentinels.

ArrayBlockingQueue is bounded and array-based. LinkedBlockingQueue may be bounded or effectively unbounded depending on construction. PriorityBlockingQueue orders elements by priority but is not bounded by a capacity parameter in the same way. DelayQueue releases elements only when their delay has expired. Selection should reflect ordering, capacity, and fairness requirements.

6. ConcurrentHashMap

ConcurrentHashMap<K,V> supports concurrent access and updates without one application-wide lock for ordinary operations. It does not permit null keys or values, because null cannot clearly represent either absence or a stored value in concurrent lookups.

ConcurrentHashMap<String, LongAdder> counts = new ConcurrentHashMap<>();
counts.computeIfAbsent("java", key -> new LongAdder()).increment();

Methods such as putIfAbsent, computeIfAbsent, compute, merge, and replace express atomic compound updates. This is safer than the check-then-act sequence:

if (!map.containsKey(key)) {
    map.put(key, value);       // unsafe compound action
}

Two separate thread-safe calls do not necessarily make the combined business operation atomic. Use the map's atomic methods or external synchronization when a larger invariant must be protected.

A ConcurrentHashMap iterator is weakly consistent: it does not generally throw ConcurrentModificationException, may reflect some updates made during traversal, and does not provide a frozen snapshot. This behavior is useful for live registries and caches but unsuitable when a stable point-in-time view is required.

LongAdder is useful for highly contended counters because it spreads updates across internal cells. Its result is appropriate for monitoring and aggregation, but it should not be treated as a transactionally exact value at every instant while updates continue.

7. Copy-on-Write Collections

CopyOnWriteArrayList creates a new internal array when the list is modified. Iterators traverse the array that existed when the iterator was created, so they are safe from concurrent modification and provide snapshot-style traversal.

CopyOnWriteArrayList<Listener> listeners = new CopyOnWriteArrayList<>();
for (Listener listener : listeners) {
    listener.onEvent(event);
}

This collection is suitable when reads and iteration are frequent but writes are rare, such as event listeners or configuration observers. Every write copies the array, making writes costly in time and memory. It is inappropriate for a frequently updated large list or for applications requiring atomic multi-operation transactions.

CopyOnWriteArraySet provides the same broad strategy with set semantics. Neither class should be selected merely because it is thread-safe; workload characteristics determine suitability.

8. Choosing Among Concurrent Structures

Thread safety is not identical to application correctness. A collection may protect its internal structure while the surrounding operation remains logically unsafe. For example, adding a value to a thread-safe list and updating a separate counter are not one atomic transaction.

9. Memory Visibility and Coordination

Concurrency correctness depends on both mutual exclusion and visibility. A write by one thread must become observable to another thread according to the Java Memory Model. Executor submission and task completion establish useful happens-before relationships: actions before submission are visible to the task, and actions in a completed task are visible after successful Future.get().

Other visibility mechanisms include synchronized, locks, volatile variables, concurrent collections, and final-field initialization rules. volatile ensures visibility and ordering for a variable, but it does not make compound operations such as increment atomic. The expression counter++ consists conceptually of read, add, and write; multiple threads can lose updates.

Prefer immutable data, confinement, message passing, and atomic collection operations. Use locks when a coherent multi-field invariant must be protected. Keep critical sections small, avoid calling unknown external code while holding locks, and establish a consistent lock order to reduce deadlock risk.

10. Illustrative Case: Document Ingestion Service

Background

A document service receives uploaded files, extracts text, records processing statistics, and notifies registered observers. Uploads arrive in bursts, while extraction may be slow and memory-intensive.

Problem

A first implementation creates a new thread for each upload and stores pending files in an unbounded list. During a burst, memory grows, database connections are exhausted, and shutdown leaves background threads alive.

Analysis

The system contains distinct workloads. Receiving and queueing are short operations. Text extraction is CPU-intensive. Database writes are blocking and must respect the connection pool. Notifications are read-heavy with infrequent listener changes. One pool for all work would allow slow database calls to starve extraction tasks.

Approach

The revised design uses:

1. a bounded ArrayBlockingQueue<Document> for admission control;
2. a fixed extraction executor sized near the available cores;
3. a separate, smaller executor for database operations;
4. ConcurrentHashMap<String, LongAdder> for per-type statistics;
5. CopyOnWriteArrayList<Listener> for notification listeners;
6. cancellation and orderly shutdown coordinated by the service owner.

A producer places documents into the bounded queue. Consumer tasks take documents and submit extraction work. Completion results are collected through ExecutorCompletionService. Successful results update statistics with merge or computeIfAbsent; failures are recorded with document identifiers and do not silently disappear.

Outcome

The queue imposes a visible capacity limit, so overload produces controlled rejection or caller back pressure rather than unlimited memory use. Separate pools prevent database latency from monopolizing CPU workers. Listener iteration remains simple and safe, while rare listener registration pays the copying cost.

Learning points

The design succeeds because concurrency mechanisms match workload properties. It also treats shutdown, monitoring, failure reporting, and downstream limits as part of correctness rather than as afterthoughts.

11. Testing, Operations, and Governance

Concurrent programs require tests that execute under different schedules. Useful techniques include repeated stress tests, barriers that deliberately align threads, randomized task delays, and assertions of invariants rather than assumptions about execution order. Tests should cover queue saturation, cancellation, interruption, task failure, executor shutdown, and duplicate processing.

Production systems should expose queue depth, active workers, task duration, rejection counts, failure counts, and cancellation rates. Thread names should identify the subsystem. Unbounded queues, accidental use of the common pool, forgotten shutdown, and swallowing InterruptedException are common operational defects.

Concurrency also has governance implications. A system that launches unlimited tasks can become a denial-of-service risk against itself or a dependent service. Access limits, database pool limits, retry budgets, and timeout policies should be explicit. Sensitive data must not be exposed in task logs merely because asynchronous processing makes tracing more difficult. Correlation identifiers help audit work across producer, queue, worker, and result stages.

12. Contemporary Java Considerations

Modern Java also supports virtual threads, which can make thread-per-task designs practical for large numbers of mostly blocking operations. Virtual threads reduce the cost of parking blocked tasks, but they do not remove limits imposed by databases, network services, CPU capacity, or memory. They should not be used to create unlimited pressure on a downstream system. CPU-bound work still requires appropriate parallelism, and synchronization or native calls can affect scalability.

The principles in this chapter remain valid: bound scarce resources, separate workloads, handle cancellation, preserve interruption, select collections according to access patterns, and measure actual behavior.

Best-Practice Checklist

• Define whether tasks are CPU-bound, blocking, periodic, or recursive.
• Prefer executors to manually creating unbounded threads.
• Give pools meaningful names and explicit ownership.
• Use bounded queues when overload must be controlled.
• Always define shutdown behavior.
• Use timeouts for operations that may wait indefinitely.
• Treat interruption as a cooperative cancellation request.
• Use atomic map methods for compound updates.
• Choose copy-on-write collections only for read-mostly workloads.
• Avoid blocking operations in Fork/Join workers unless deliberately managed.
• Monitor queue length, latency, failures, and rejection.
• Test under saturation and unusual scheduling.

Chapter Summary

Executors provide a structured way to submit work without coupling application logic to thread creation. ExecutorService manages task execution and lifecycle; Future represents delayed results and supports timeout and cancellation; CompletionService enables completion-order processing. Fixed, cached, scheduled, and work-stealing pools serve different workload patterns, while bounded ThreadPoolExecutor configurations provide back pressure.

Fork/Join is effective for sufficiently large, independent, CPU-bound divide-and-conquer computations. BlockingQueue supports safe producer-consumer pipelines and can limit overload. ConcurrentHashMap offers scalable concurrent map operations and atomic methods for common compound updates. Copy-on-write collections provide safe snapshot-style traversal when writes are rare. None of these tools automatically makes a complete business operation atomic; visibility, invariants, cancellation, shutdown, and external resource limits must also be designed.

Key Terms

Executor: An object that accepts tasks for execution.

ExecutorService: An executor with task submission, result, and lifecycle management operations.

Callable: A task that returns a value and may throw an exception.

Future: A handle representing the eventual result or status of an asynchronous task.

Thread pool: A reusable group of worker threads that execute submitted tasks.

Back pressure: A mechanism that slows or rejects producers when consumers or capacity are constrained.

BlockingQueue: A queue whose insertion or removal operations may wait for capacity or data.

Producer-consumer pattern: A design in which producers create work and consumers process queued work.

ConcurrentHashMap: A concurrent key-value map supporting scalable access and atomic update methods.

Copy-on-write: A strategy that copies shared data when it is modified, enabling stable read traversal.

Fork/Join: A framework for recursively splitting and joining parallel computations.

RecursiveTask: A Fork/Join task that returns a result.

Work stealing: A scheduling technique in which idle workers obtain tasks from other workers.

Weakly consistent iterator: An iterator that tolerates concurrent updates without providing a fixed snapshot.

Memory visibility: The property that updates by one thread become observable by another under defined synchronization rules.

Interruption: A cooperative signal requesting that a thread stop waiting or terminate.

Atomic compound action: A multi-step operation performed as one indivisible logical action.

Bounded queue: A queue with a defined maximum capacity.

LongAdder: A scalable counter designed for updates under high contention.

Review Questions

Short Answer Questions

1. What is the primary advantage of using ExecutorService instead of creating a new thread for every task?
2. Distinguish Runnable from Callable.
3. What happens when BlockingQueue.put is called on a full bounded queue?
4. Why is CopyOnWriteArrayList suitable for read-mostly workloads?
5. Why should interruption status usually be restored after catching InterruptedException?

Long Answer Questions

1. Explain the lifecycle of an executor, including task submission, shutdown, termination, and interruption.
2. Compare fixed, cached, scheduled, and work-stealing pools, including suitable use cases and risks.
3. Explain how ConcurrentHashMap supports concurrent updates and why check-then-act code can remain unsafe.
4. Describe the Fork/Join model and explain why task granularity and blocking behavior affect performance.
5. Design a producer-consumer pipeline using a bounded queue, multiple consumers, failure handling, and orderly shutdown.

Analytical / Application Questions

1. A service uses one unbounded fixed thread pool for CPU-intensive parsing and slow database writes. Analyze the likely failure modes and propose a revised executor architecture.
2. A shared map records whether a customer has already been processed. Explain why containsKey followed by put may duplicate work and show how an atomic map operation could improve the design.
3. A notification system has thousands of event reads per minute but listener registration occurs only a few times per day. Select an appropriate collection, justify the choice, and identify its principal limitation.

Multiple Choice Questions with Answers

Question 1

Which interface is most appropriate for a task that returns a value and may throw a checked exception?

A. Runnable

B. Callable

C. Executor

D. Thread

Correct Answer: B. Callable

Question 2

What does shutdown() do to an ExecutorService?

A. Immediately interrupts every running task

B. Deletes all queued tasks

C. Rejects new tasks while allowing submitted tasks to complete

D. Restarts workers with a larger pool

Correct Answer: C. Rejects new tasks while allowing submitted tasks to complete

Question 3

Which structure is specifically designed to wait when a queue is full or empty?

A. HashMap

B. ArrayList

C. BlockingQueue

D. TreeSet

Correct Answer: C. BlockingQueue

Question 4

Which method is appropriate for an atomic absent-key initialization in ConcurrentHashMap?

A. containsKey followed by get

B. computeIfAbsent

C. size followed by put

D. keySet followed by remove

Correct Answer: B. computeIfAbsent

Question 5

Which workload is the best fit for Fork/Join?

A. A recursively divisible CPU-bound calculation

B. A single blocking database transaction

C. A continuously growing log queue

D. A listener list with rare writes

Correct Answer: A. A recursively divisible CPU-bound calculation

Question 6

What is a principal cost of CopyOnWriteArrayList?

A. Reads always require a lock

B. Every modification copies the underlying array

C. It cannot be safely iterated

D. It does not preserve element order

Correct Answer: B. Every modification copies the underlying array

Question 7

What is the purpose of a bounded executor queue?

A. To guarantee that no task ever waits

B. To provide unlimited task storage

C. To impose capacity and support overload control

D. To disable task cancellation

Correct Answer: C. To impose capacity and support overload control

Question 8

What does Future.get with a timeout provide?

A. A guarantee that the task finishes within the timeout

B. A bounded wait by the calling thread

C. Automatic retry after failure

D. A larger worker pool

Correct Answer: B. A bounded wait by the calling thread

Question 9

Why can counter++ be unsafe when shared by multiple threads?

A. It is a compound read-modify-write operation

B. It always throws an exception

C. It prevents memory visibility

D. It creates a new executor

Correct Answer: A. It is a compound read-modify-write operation

Question 10

Which collection is generally suitable for a high-contention monitoring counter?

A. LongAdder

B. CopyOnWriteArrayList

C. PriorityBlockingQueue

D. TreeMap

Correct Answer: A. LongAdder