Jobs that run at a scheduled time, ordered by priority, pulled by a pool of worker threads. A PriorityQueue, a ReentrantLock, and one genuinely new primitive — the timed wait — are all it takes.

This builds directly on the bounded blocking queue: same producer–consumer skeleton, two new twists. If that post is fresh, this one will click fast.

The problem

Build a scheduler that accepts jobs, each with a scheduled time and a priority. A job runs when its time arrives; if two jobs share a time, the higher-priority one runs first. Many threads submit and consume concurrently.

This is the producer–consumer problem with two twists. Producers submit jobs; a pool of worker threads runs them. But unlike a plain queue, jobs come out in a specific order — by scheduled time, then priority — and a worker must wait until a job's time arrives before running it. Those two twists, ordering and time-gating, are what make it more than a blocking queue.

Java's ScheduledThreadPoolExecutor does this in production. But building it by hand forces you to understand priority ordering, timed waiting, and the asymmetry of an unbounded producer–consumer system — all common interview ground.

The shape of the solution

Every problem in this family — bounded queue, scheduler, thread pool — has the same skeleton. Five questions generate the design:

  • What's the shared state? A PriorityQueue of jobs, ordered by time then priority.
  • Who are the actors? Submitters (add jobs) and workers (run due jobs). So: submit() and take().
  • When does each wait? A submitter never waits (unbounded — the queue never fills). A worker waits when the queue is empty, or when the earliest job isn't due yet.
  • What unblocks a waiter? A worker waiting is unblocked by a submit adding a job (possibly sooner than what it's waiting for). So submit() signals; take() signals no one, because no producer ever waits.
  • What's the mechanism? One ReentrantLock guarding the queue, one Condition where workers wait.

The asymmetry to notice

A bounded queue is symmetric: both producers and consumers wait, and both signal. An unbounded scheduler is asymmetric — only workers wait (for a job to be ready), and only submit() signals (to wake them). Since the queue never fills, producers never block, so take() removing a job unblocks nobody and needs no signal. Recognising this saves you from writing signalling that does nothing.

The Job and its ordering

A job is a passive unit of work: it holds what to do, when, and at what priority. Crucially, its run() executes synchronously — on whatever worker thread calls it. The job doesn't spawn its own thread; the worker pool owns the threading.

class Job {
    int id;
    Priority priority;
    long runAtMillis;              // when it should run

    Job(int id, Priority priority, long runAtMillis) {
        this.id = id;
        this.priority = priority;
        this.runAtMillis = runAtMillis;
    }

    void run() {
        System.out.println("Job " + id + " ran at " + System.currentTimeMillis());
    }
}

enum Priority {
    HIGH(1), MEDIUM(2), LOW(3);   // lower level = higher priority
    private final int level;
    Priority(int level) { this.level = level; }
    public int getLevel() { return level; }
}

Why an explicit level in the enum

The enum carries an explicit int level rather than relying on ordinal(). Ordinal depends on declaration order, so reordering the constants would silently change the values. An explicit field assigned in the constructor is stable and deliberate — the number means what you say it means, regardless of declaration order.

PriorityQueue and the comparator

A PriorityQueue is a queue that always hands you the "smallest" element first, by a comparator you supply. Its API mirrors a plain queue — offer, poll, peek, size — the only difference being that poll returns the smallest rather than the oldest.

For a scheduler, "smallest" means "should run soonest": earliest time first, and among ties, highest priority.

PriorityQueue<Job> pq = new PriorityQueue<>((a, b) -> {
    if (a.runAtMillis != b.runAtMillis) {
        return Long.compare(a.runAtMillis, b.runAtMillis);   // primary: earliest time
    }
    return Integer.compare(a.priority.getLevel(), b.priority.getLevel()); // tie-break: priority
});

Why compare returns an int

A comparator returns an int because ordering is a three-way answer — before, equal, or after — and the int's sign encodes which: negative means the first element sorts first, zero means equal, positive means it sorts after. Only the sign matters, never the magnitude, which is why Long.compare (returning −1/0/+1) works and is safer than a - b, which can overflow.

The one PriorityQueue template to remember

A lambda comparator with compare inside: (a, b) -> Integer.compare(a.field, b.field) for ascending, swap to (b, a) for descending, and an if-block for multi-field sorting. Same shape scales from one field to many. Use Integer.compare/Long.compare, never a - b.

submit — the easy half

Because the scheduler is unbounded, a submitter never waits. submit() is simply: lock, add the job, wake the workers, unlock. No while, no waiting.

void submit(Job job) {
    lock.lock();
    try {
        pq.offer(job);
        available.signalAll();   // wake workers — a job is now available
    } finally {
        lock.unlock();
    }
}

The signalAll() matters even though the queue was just added to: a worker might be sleeping on a timer for a later job, and this new job could be sooner. Waking all workers lets each re-check the head of the queue and re-wait for the correct, possibly shorter, delay.

Why signalAll, not signal?

A fair question, and a favourite interview follow-up: submit() adds exactly one job, so why wake all the workers with signalAll() instead of just one with signal()?

Because the workers parked on available are not interchangeable. They're waiting in two different states on the same condition:

  • empty queue → await() (indefinite — "wake me when any job arrives")
  • head not due → awaitNanos(delay) (timed — "wake me at this instant, or if something changes")

signal() wakes one arbitrary waiter. The trap is that the woken worker can re-check the head, find it still not due, and go straight back to awaitNanos() — swallowing the wakeup — while a worker that could have done real work is never woken.

The scenario signal() breaks

Two workers, A and B. (1) Queue empty — both park in await(). (2) Submit J1, due in 3s — signal() wakes A, which then sleeps in awaitNanos(3s); B stays in indefinite await(). (3) Submit J2, also due in 3s — signal() wakes A again; A peeks the head, sees it's still 3s out, and re-sleeps, absorbing the signal. B is never woken. (4) At +3s, A wakes and runs J1, then J2 — sequentially. If J1 takes 4s, J2 fires ~4s late, even though B sat idle and available the whole time.

No job is lost, but jobs that should run in parallel at their due time get serialised — and a job can miss its scheduled time stuck behind another's execution. For a scheduler, that's a real correctness bug, not just a throughput dent. signalAll() wakes both workers; each re-evaluates the head, and the second one picks up J2, so the two run together.

The rule

signal() is safe only when every waiter is waiting for the identical condition and one wakeup lets exactly one thread proceed. Here the waiters sit in mixed indefinite/timed states on one condition — so waking one arbitrary thread can strand work, and signalAll() is the safe default. Its only cost is mild: every submit wakes all workers, and the ones that can't proceed simply re-check the head and wait again.

The timed wait — the one new primitive

The bounded queue used plain await() — wait indefinitely until signalled. A scheduler can't use that alone, and there's a specific reason why.

The bug plain await() would cause

Suppose a job is scheduled ten seconds out. A worker picks it up, sees it's not due, and calls plain await(). Now nothing wakes it — time passing doesn't call signal(). Unless another job happens to be submitted, the worker sleeps forever and the scheduled job never runs, even though its time arrives. A job becoming due is a function of time, not of any signal — so waiting for a signal alone misses it.

The fix is a timed wait: wait until signalled or until a deadline, whichever comes first. Condition offers several:

Method Wakes when
await() signalled only — indefinite
awaitNanos(nanos) signalled OR nanos elapse
await(time, unit) signalled OR time elapses
awaitUntil(date) signalled OR the date arrives

The distinction that solves scheduling

Use a timed wait when a worker is waiting for a job's time to arrive — the timer wakes it when the job is due, with no signal needed. Use plain await() only when the queue is empty, where there's no time to wait for, only a future submit. So: empty queue → await(); job not due yet → awaitNanos(delay). The timer fires the job on time; the signal handles a newly-arrived sooner job jumping ahead.

take — the hard half

A worker calls take() to get the next due job. It handles three cases in a loop: empty (wait indefinitely), head not due (wait until its time), head due (take it and return).

Job take() throws InterruptedException {
    Job job;
    lock.lock();
    try {
        while (true) {
            if (pq.isEmpty()) {
                available.await();                          // empty → wait for a submit
            } else {
                Job head = pq.peek();                       // LOOK, don't remove
                long delay = head.runAtMillis - System.currentTimeMillis();
                if (delay <= 0) {
                    job = pq.poll();                        // due → remove it
                    break;                                  // exit loop to run outside the lock
                } else {
                    available.awaitNanos(delay * 1_000_000L); // not due → wait until its time
                }
            }
        }
    } finally {
        lock.unlock();
    }
    job.run();   // run OUTSIDE the lock
    return job;
}

Three details carry the correctness:

peek to decide, poll to commit

You peek() the head to read its scheduled time without removing it. Only once you confirm it's due do you poll(). Polling before the due-check would remove a not-due job and then leave it stranded in a local variable when you wait — the job would vanish from the queue. Look first; remove only when committing to run.

break on the success path

The while(true) is a waiting loop — it loops while the worker can't proceed (empty, or not due). The moment it has a due job, it breaks out. Without the break, the loop would spin forever and job.run() would be unreachable. The waiting branches loop back; the got-it branch breaks.

run outside the lock

The job is extracted under the lock but run after releasing it. Running inside the lock would hold it for the entire job execution, blocking every other worker and every submit. Extract under the lock, break, unlock via finally, then run — so job execution never blocks the rest of the scheduler.

Bounded vs unbounded — a design decision

Before coding, it's worth surfacing: should the scheduler be bounded (cap the queue, apply backpressure) or unbounded (accept any number of jobs)? The two behave differently:

Unbounded Bounded
Producer waits? Never — queue never fills Yes — when full
submit signals? Yes — wakes workers Yes — wakes workers
take signals? No — no producer waits Yes — wakes producers waiting for space
Conditions one (workers) two (notFull, notEmpty)
Risk unbounded memory growth backpressure, more code

The interview move

Default to unbounded — it's the standard meaning of "scheduler" and keeps focus on the timing logic that actually demonstrates skill. But raise the question first: "should this apply backpressure past some capacity, or accept unlimited jobs? I'll assume unbounded unless you want a bound." Surfacing the trade-off reads as senior-level; silently picking one doesn't. And if they want it bounded, it's exactly the bounded-blocking-queue pattern — a notFull condition, submit waits when full, take signals notFull after removing.

Workers and the two loops

The subtlety that trips people: there are two while loops, at different levels, doing different jobs.

// WORKER THREAD — its whole life
new Thread(() -> {
    while (true) {                    // Loop 2: keep getting jobs, FOREVER
        try {
            scheduler.take();         // Loop 1 runs inside — waits for and runs ONE job
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            break;
        }
    }
}).start();

Two loops, two jobs

Loop 1, inside take(), waits until one job is ready, runs it, and returns — so take() handles a single job per call. Loop 2, in the worker, calls take() repeatedly, forever — so one worker processes many jobs over its lifetime. Without Loop 2, a worker would take one job, run it, and end. Returning from take() doesn't kill the thread — the thread lives in Loop 2, which sends it back to take() for the next job.

How the signal and the loop cooperate

A producer's signalAll() only wakes a worker that's currently blocked inside take(). Loop 2 is what keeps returning the worker to that blocked state after each job — so it's there to be woken by the next signal. The signal ends one wait; the loop re-arms the worker for the next. Neither alone suffices: the signal is the wakeup, the loop keeps the worker in position to be woken.

Simulating it

To exercise the scheduler, spawn several workers and several producers. Give jobs different times and priorities so the ordering is observable, use daemon workers so the JVM can exit, and sleep in main long enough for the jobs to fire.

public static void main(String[] args) throws InterruptedException {
    JobScheduler scheduler = new JobScheduler();
    long now = System.currentTimeMillis();

    for (int c = 0; c < 3; c++) {                  // 3 workers, share the load
        Thread t = new Thread(() -> {
            while (true) {
                try { scheduler.take(); }
                catch (InterruptedException e) { Thread.currentThread().interrupt(); break; }
            }
        });
        t.setDaemon(true);                          // JVM can exit while these loop
        t.start();
    }

    for (int p = 0; p < 4; p++) {                  // 4 producers, a few jobs each
        final int pid = p;
        new Thread(() -> {
            for (int j = 0; j < 3; j++) {
                scheduler.submit(new Job(pid * 10 + j, Priority.HIGH, now + (j + 1) * 1000L));
            }
        }).start();
    }

    Thread.sleep(5000);                            // let jobs (within 3s) run, then exit
}

Workers run forever, so daemon threads plus a sleep give a clean demo exit. In production you'd use an explicit shutdown flag rather than setDaemon.

The complete program

Here is the whole thing in one file — the Priority enum, the Job, the JobScheduler, and a main that varies priority across producers so you can watch both the time-ordering and the priority tie-break. Save it as JobSchedulerDemo.java and run it with a single command (JDK 11+).

import java.util.PriorityQueue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

enum Priority {
    HIGH(1), MEDIUM(2), LOW(3);          // lower level = higher priority
    private final int level;
    Priority(int level) { this.level = level; }
    public int getLevel() { return level; }
}

class Job {
    final int id;
    final Priority priority;
    final long runAtMillis;              // when it should run

    Job(int id, Priority priority, long runAtMillis) {
        this.id = id;
        this.priority = priority;
        this.runAtMillis = runAtMillis;
    }

    void run() {
        System.out.println("ran job " + id + " (" + priority + ")");
    }
}

class JobScheduler {
    private final PriorityQueue<Job> pq = new PriorityQueue<>((a, b) -> {
        if (a.runAtMillis != b.runAtMillis) {
            return Long.compare(a.runAtMillis, b.runAtMillis);            // primary: earliest time
        }
        return Integer.compare(a.priority.getLevel(), b.priority.getLevel()); // tie-break: priority
    });
    private final ReentrantLock lock = new ReentrantLock();
    private final Condition available = lock.newCondition();             // workers wait here

    void submit(Job job) {
        lock.lock();
        try {
            pq.offer(job);
            available.signalAll();       // wake workers — a job is now available
        } finally {
            lock.unlock();
        }
    }

    Job take() throws InterruptedException {
        Job job;
        lock.lock();
        try {
            while (true) {
                if (pq.isEmpty()) {
                    available.await();                          // empty → wait for a submit
                } else {
                    Job head = pq.peek();                       // LOOK, don't remove
                    long delay = head.runAtMillis - System.currentTimeMillis();
                    if (delay <= 0) {
                        job = pq.poll();                        // due → remove it
                        break;                                  // exit loop to run outside the lock
                    } else {
                        available.awaitNanos(delay * 1_000_000L); // not due → wait until its time
                    }
                }
            }
        } finally {
            lock.unlock();
        }
        job.run();   // run OUTSIDE the lock
        return job;
    }
}

public class JobSchedulerDemo {
    public static void main(String[] args) throws InterruptedException {
        JobScheduler scheduler = new JobScheduler();
        long now = System.currentTimeMillis();

        for (int c = 0; c < 3; c++) {                  // 3 workers, share the load
            Thread t = new Thread(() -> {
                while (true) {
                    try { scheduler.take(); }
                    catch (InterruptedException e) { Thread.currentThread().interrupt(); break; }
                }
            });
            t.setDaemon(true);                          // JVM can exit while these loop
            t.start();
        }

        for (int p = 0; p < 4; p++) {                  // 4 producers, a few jobs each
            final int pid = p;
            Priority pr = Priority.values()[pid % 3];  // vary priority so tie-breaks show
            new Thread(() -> {
                for (int j = 0; j < 3; j++) {
                    scheduler.submit(new Job(pid * 10 + j, pr, now + (j + 1) * 1000L));
                }
            }).start();
        }

        Thread.sleep(5000);                            // let jobs (within 3s) run, then exit
    }
}

Run it:

java JobSchedulerDemo.java

The twelve jobs fire in three waves one second apart — every job with j == 0 at +1s, then j == 1 at +2s, then j == 2 at +3s — proof the time-gating works. Within a wave the exact print order interleaves across the three workers (they poll near-simultaneously), but each wave's jobs all share a scheduled second, and the program always exits cleanly after ~5s:

ran job 30 (HIGH)
ran job 20 (LOW)
ran job 10 (MEDIUM)
ran job 0 (HIGH)
ran job 31 (HIGH)
ran job 1 (HIGH)
ran job 11 (MEDIUM)
ran job 21 (LOW)
ran job 2 (HIGH)
ran job 32 (HIGH)
ran job 12 (MEDIUM)
ran job 22 (LOW)

To see strict time-then-priority order — HIGH before MEDIUM before LOW within each wave — drop to a single worker; with one consumer, poll() hands jobs out in exact comparator order.

Revision cheat sheet

The scheduler, distilled — read this to reconstruct the whole thing.

The five design questions

  • Shared state: a PriorityQueue ordered by time, then priority.
  • Actors: submitters (submit) and workers (take).
  • When wait: submitter never (unbounded); worker when empty OR head not due.
  • What unblocks: a submit adding a job wakes workers; take signals no one (no producer waits).
  • Mechanism: one ReentrantLock, one Condition.

PriorityQueue

  • Same API as Queue (offer/poll/peek/size) but poll returns the smallest.
  • Template: (a, b) -> Integer.compare(a.field, b.field); if-block for multi-field. Use Long/Integer.compare, never a - b (overflow).
  • compare returns an int because ordering is three-way; only the sign matters.

The timed wait — the key idea

  • A job becomes due because time passed, not because of a signal. So plain await() would leave a scheduled job unrun if no other job arrived.
  • Empty queue → await() (indefinite — nothing to time). Head not due → awaitNanos(delay) (wake at its time, or on a sooner job's signal).

take() — the three must-gets

  • peek to decide, poll to commit — never remove a job until you've confirmed it's due, or a not-due job vanishes.
  • break on success — the while(true) waits; it breaks the moment it has a due job, so run() is reachable.
  • run outside the lock — extract under the lock, break, unlock, then run, so execution doesn't block other workers.

signalAll vs signal

  • Workers wait on one condition in two states — empty (await()) and not-due (awaitNanos). They're not interchangeable.
  • signal() can wake a worker that just re-sleeps on its timer, swallowing the wakeup and stranding an idle worker → jobs serialise and can miss their scheduled time.
  • signalAll() is the safe default; its only cost is waking workers that re-check the head and wait again.

Threads — the two loops

  • Loop 1 (inside take): wait for one job, run it, return. One job per call.
  • Loop 2 (the worker): call take forever. Processes many jobs over its life.
  • Returning from take doesn't kill the thread — it lives in Loop 2, which sends it back. A signal only wakes a worker blocked inside take; Loop 2 keeps returning it there.
  • Unbounded asymmetry: only workers wait, only submit signals.

The scheduler is your bounded blocking queue plus two twists — priority ordering and time-gated waiting. The one genuinely new primitive is the timed wait; everything else is the producer–consumer skeleton you already know. If you spot an error or a cleaner framing, I'd like to hear it.