Synchronization asks: who may access this state now?
Coordination asks: who should do which work, when, and under what capacity limit?

final class AuditEventProcessor implements AutoCloseable {
    private final BlockingQueue<AuditEvent> queue;
    private final ExecutorService workers;
    private final AtomicBoolean running = new AtomicBoolean(true);

    AuditEventProcessor(int capacity, int workerCount) {
        this.queue = new ArrayBlockingQueue<>(capacity);
        this.workers = Executors.newFixedThreadPool(workerCount);

        for (int i = 0; i < workerCount; i++) {
            workers.submit(this::consumeLoop);
        }
    }

    boolean publish(AuditEvent event, Duration timeout) throws InterruptedException {
        if (!running.get()) {
            return false;
        }
        return queue.offer(event, timeout.toMillis(), TimeUnit.MILLISECONDS);
    }

    private void consumeLoop() {
        while (running.get() || !queue.isEmpty()) {
            try {
                AuditEvent event = queue.poll(500, TimeUnit.MILLISECONDS);
                if (event != null) {
                    persist(event);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            } catch (Exception e) {
                // In production: classify, retry, dead-letter, or alert.
                System.err.println("failed to process audit event: " + e.getMessage());
            }
        }
    }

    private void persist(AuditEvent event) {
        // database write
    }

    @Override
    public void close() {
        running.set(false);
        workers.shutdown();
    }
}

Every accepted item is either processed, explicitly rejected, retried, quarantined, or durably recoverable.

BlockingQueue<Job> queue = new ArrayBlockingQueue<>(10_000);

boolean accepted = queue.offer(job, 100, TimeUnit.MILLISECONDS);
if (!accepted) {
    throw new RejectedExecutionException("job queue is full");
}

final class BoundedExecutor implements AutoCloseable {
    private final ThreadPoolExecutor executor;

    BoundedExecutor(int threads, int queueCapacity, String namePrefix) {
        BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(queueCapacity);
        ThreadFactory factory = Thread.ofPlatform()
                .name(namePrefix + "-", 0)
                .factory();

        this.executor = new ThreadPoolExecutor(
                threads,
                threads,
                0L,
                TimeUnit.MILLISECONDS,
                queue,
                factory,
                new ThreadPoolExecutor.AbortPolicy()
        );
    }

    Future<?> submit(Runnable task) {
        return executor.submit(task);
    }

    @Override
    public void close() {
        executor.shutdown();
    }
}

ExecutorService executor = Executors.newFixedThreadPool(8);

threads ≈ number of available processors

threads may need to exceed CPU count, but must be bounded by downstream capacity

do not pool virtual threads for CPU sizing;
limit scarce resources with semaphores, connection pools, rate limiters, or structured concurrency

final class SumTask extends RecursiveTask<Long> {
    private static final int THRESHOLD = 10_000;

    private final long[] values;
    private final int start;
    private final int end;

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

    @Override
    protected Long compute() {
        int length = end - start;
        if (length <= THRESHOLD) {
            long sum = 0;
            for (int i = start; i < end; i++) {
                sum += values[i];
            }
            return sum;
        }

        int mid = start + length / 2;
        SumTask left = new SumTask(values, start, mid);
        SumTask right = new SumTask(values, mid, end);

        left.fork();
        long rightResult = right.compute();
        long leftResult = left.join();

        return leftResult + rightResult;
    }
}

long result = ForkJoinPool.commonPool().invoke(new SumTask(values, 0, values.length));

final class CaseSummaryService {
    private final ExecutorService executor;

    CaseSummaryService(ExecutorService executor) {
        this.executor = executor;
    }

    CaseSummary load(CaseId id) throws InterruptedException, ExecutionException {
        Future<CaseFile> file = executor.submit(() -> loadFile(id));
        Future<RiskScore> risk = executor.submit(() -> loadRisk(id));
        Future<List<Action>> actions = executor.submit(() -> loadActions(id));

        return new CaseSummary(file.get(), risk.get(), actions.get());
    }
}

CaseSummary load(CaseId id, Duration timeout) throws Exception {
    List<Future<?>> futures = new ArrayList<>();
    try {
        Future<CaseFile> file = executor.submit(() -> loadFile(id));
        Future<RiskScore> risk = executor.submit(() -> loadRisk(id));
        Future<List<Action>> actions = executor.submit(() -> loadActions(id));

        futures.add(file);
        futures.add(risk);
        futures.add(actions);

        long deadline = System.nanoTime() + timeout.toNanos();

        return new CaseSummary(
                getBefore(file, deadline),
                getBefore(risk, deadline),
                getBefore(actions, deadline)
        );
    } catch (Exception e) {
        for (Future<?> future : futures) {
            future.cancel(true);
        }
        throw e;
    }
}

private static <T> T getBefore(Future<T> future, long deadlineNanos) throws Exception {
    long remaining = deadlineNanos - System.nanoTime();
    if (remaining <= 0) {
        throw new TimeoutException("deadline exceeded");
    }
    return future.get(remaining, TimeUnit.NANOSECONDS);
}

ExecutorCompletionService<Result> completion = new ExecutorCompletionService<>(executor);

for (Input input : inputs) {
    completion.submit(() -> process(input));
}

List<Result> results = new ArrayList<>();
for (int i = 0; i < inputs.size(); i++) {
    Future<Result> completed = completion.take();
    results.add(completed.get());
}

Future<Result> completed = completion.poll(200, TimeUnit.MILLISECONDS);
if (completed == null) {
    throw new TimeoutException("no task completed in time");
}

CountDownLatch ready = new CountDownLatch(workerCount);
CountDownLatch start = new CountDownLatch(1);
CountDownLatch done = new CountDownLatch(workerCount);

for (int i = 0; i < workerCount; i++) {
    executor.submit(() -> {
        ready.countDown();
        try {
            start.await();
            doWork();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            done.countDown();
        }
    });
}

ready.await();      // all workers ready
start.countDown();  // release them together
done.await();       // wait completion

CyclicBarrier barrier = new CyclicBarrier(workerCount, () -> {
    System.out.println("phase completed");
});

Runnable worker = () -> {
    try {
        while (!Thread.currentThread().isInterrupted()) {
            computePhase();
            barrier.await();
            applyPhaseResult();
            barrier.await();
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    } catch (BrokenBarrierException e) {
        // another participant failed or barrier was reset
    }
};

Phaser phaser = new Phaser(1); // main party

for (Task task : tasks) {
    phaser.register();
    executor.submit(() -> {
        try {
            doPhaseOne(task);
            phaser.arriveAndAwaitAdvance();

            doPhaseTwo(task);
            phaser.arriveAndAwaitAdvance();
        } finally {
            phaser.arriveAndDeregister();
        }
    });
}

phaser.arriveAndDeregister(); // main deregisters

final class ReportLimiter {
    private final Semaphore semaphore = new Semaphore(5);

    Report generate(ReportRequest request) throws InterruptedException {
        if (!semaphore.tryAcquire(1, TimeUnit.SECONDS)) {
            throw new RejectedExecutionException("report capacity exhausted");
        }
        try {
            return generateReport(request);
        } finally {
            semaphore.release();
        }
    }
}

final class TenantLimiter {
    private final ConcurrentHashMap<TenantId, Semaphore> semaphores = new ConcurrentHashMap<>();
    private final int permitsPerTenant;

    TenantLimiter(int permitsPerTenant) {
        this.permitsPerTenant = permitsPerTenant;
    }

    <T> T withPermit(TenantId tenantId, Callable<T> action) throws Exception {
        Semaphore semaphore = semaphores.computeIfAbsent(
                tenantId,
                ignored -> new Semaphore(permitsPerTenant)
        );

        if (!semaphore.tryAcquire(500, TimeUnit.MILLISECONDS)) {
            throw new RejectedExecutionException("tenant capacity exhausted");
        }
        try {
            return action.call();
        } finally {
            semaphore.release();
        }
    }
}

final class ExecutorShutdown {
    static void shutdownGracefully(ExecutorService executor, Duration timeout) {
        executor.shutdown();
        try {
            if (!executor.awaitTermination(timeout.toMillis(), TimeUnit.MILLISECONDS)) {
                List<Runnable> dropped = executor.shutdownNow();
                System.err.println("forced shutdown; dropped tasks=" + dropped.size());

                if (!executor.awaitTermination(timeout.toMillis(), TimeUnit.MILLISECONDS)) {
                    System.err.println("executor did not terminate");
                }
            }
        } catch (InterruptedException e) {
            executor.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }
}

while (!Thread.currentThread().isInterrupted()) {
    Task task = queue.poll(500, TimeUnit.MILLISECONDS);
    if (task != null) {
        process(task);
    }
}

enum WorkItem {
    POISON
}

sealed interface WorkItem permits RealWork, StopWork {}
record RealWork(String payload) implements WorkItem {}
record StopWork() implements WorkItem {}

void consumeLoop(BlockingQueue<WorkItem> queue) throws InterruptedException {
    while (true) {
        WorkItem item = queue.take();
        if (item instanceof StopWork) {
            return;
        }
        process((RealWork) item);
    }
}

final class WorkBulkheads implements AutoCloseable {
    private final ExecutorService caseTransitions = Executors.newFixedThreadPool(16);
    private final ExecutorService reports = Executors.newFixedThreadPool(4);
    private final ExecutorService notifications = Executors.newFixedThreadPool(8);

    Future<?> submitCaseTransition(Runnable task) {
        return caseTransitions.submit(task);
    }

    Future<?> submitReport(Runnable task) {
        return reports.submit(task);
    }

    Future<?> submitNotification(Runnable task) {
        return notifications.submit(task);
    }

    @Override
    public void close() {
        caseTransitions.shutdown();
        reports.shutdown();
        notifications.shutdown();
    }
}

record PrioritizedTask(int priority, long sequence, Runnable task)
        implements Comparable<PrioritizedTask> {

    @Override
    public int compareTo(PrioritizedTask other) {
        int byPriority = Integer.compare(other.priority, this.priority);
        if (byPriority != 0) {
            return byPriority;
        }
        return Long.compare(this.sequence, other.sequence);
    }
}

ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);

ScheduledFuture<?> future = scheduler.scheduleAtFixedRate(
        () -> refreshCacheSafely(),
        0,
        30,
        TimeUnit.SECONDS
);

void refreshCacheSafely() {
    try {
        refreshCache();
    } catch (Exception e) {
        // never let scheduled task die silently
        System.err.println("cache refresh failed: " + e.getMessage());
    }
}

ExecutorService pool = Executors.newFixedThreadPool(2);

Callable<String> outer = () -> {
    Future<String> inner = pool.submit(() -> "inner");
    return inner.get();
};

pool.submit(outer);
pool.submit(outer);

CompletableFuture.supplyAsync(() -> slowBlockingHttpCall());

CompletableFuture.supplyAsync(
        () -> slowBlockingHttpCall(),
        blockingIoExecutor
);

final class PartitionedExecutor implements AutoCloseable {
    private final ExecutorService[] executors;

    PartitionedExecutor(int partitions) {
        this.executors = new ExecutorService[partitions];
        for (int i = 0; i < partitions; i++) {
            executors[i] = Executors.newSingleThreadExecutor(
                    Thread.ofPlatform().name("partition-" + i + "-", 0).factory()
            );
        }
    }

    Future<?> submit(String key, Runnable task) {
        return executorFor(key).submit(task);
    }

    private ExecutorService executorFor(String key) {
        int index = Math.floorMod(key.hashCode(), executors.length);
        return executors[index];
    }

    @Override
    public void close() {
        for (ExecutorService executor : executors) {
            executor.shutdown();
        }
    }
}

Tasks for the same key execute sequentially.
Tasks for different keys may execute concurrently.

final class EmailWorker {
    private final ProcessedMessageRepository processed;
    private final EmailGateway gateway;

    void handle(EmailJob job) {
        if (processed.exists(job.id())) {
            return;
        }

        gateway.send(job.recipient(), job.subject(), job.body());
        processed.markProcessed(job.id());
    }
}

insert into processed_message(message_id, processed_at)
values (?, current_timestamp)

executor.active_threads
executor.pool_size
executor.queue_depth
executor.completed_tasks
executor.rejected_tasks
executor.task_duration
executor.task_queue_age
worker.failures_by_type
worker.retries
worker.dead_letters
shutdown.drain_duration

event=task_rejected queue=case-transition reason=queue_full tenant=...
event=task_timeout taskType=report durationMs=...
event=worker_failure taskId=... exceptionClass=...
event=shutdown_forced droppedTasks=...

@Test
void rejectsWhenQueueFull() throws Exception {
    BlockingQueue<Job> queue = new ArrayBlockingQueue<>(1);
    queue.put(new Job("existing"));

    boolean accepted = queue.offer(new Job("new"), 10, TimeUnit.MILLISECONDS);

    assertFalse(accepted);
}

@Test
void workersStartTogether() throws Exception {
    int workers = 10;
    CountDownLatch ready = new CountDownLatch(workers);
    CountDownLatch start = new CountDownLatch(1);
    CountDownLatch done = new CountDownLatch(workers);
    AtomicInteger counter = new AtomicInteger();

    ExecutorService executor = Executors.newFixedThreadPool(workers);
    try {
        for (int i = 0; i < workers; i++) {
            executor.submit(() -> {
                ready.countDown();
                try {
                    start.await();
                    counter.incrementAndGet();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally {
                    done.countDown();
                }
            });
        }

        assertTrue(ready.await(1, TimeUnit.SECONDS));
        start.countDown();
        assertTrue(done.await(1, TimeUnit.SECONDS));
        assertEquals(workers, counter.get());
    } finally {
        executor.shutdownNow();
    }
}

- events for the same case must be processed in order
- events for different cases may run concurrently
- queue must be bounded
- failed event must not silently disappear
- shutdown should drain accepted events for up to 30 seconds

final class NotificationService {
    private final ExecutorService executor = Executors.newFixedThreadPool(20);

    void notifyLater(Notification notification) {
        executor.submit(() -> send(notification));
    }
}

Which executor should I use?

What work exists?
Who owns it?
How is it bounded?
How does it fail?
How is it cancelled?
How is it observed?
How does it shut down?
What invariant must not be broken by parallelism?