Building a Multi‑Priority Thread Pool with Java's PriorityBlockingQueue

In earlier Java thread‑pool experiments the author faced task‑priority issues and initially used two pools, but that approach became complex.

After revisiting java.util.concurrent, the optimal solution is to use java.util.concurrent.PriorityBlockingQueue.

PriorityBlockingQueue Overview

java.util.concurrent.PriorityBlockingQueue is a thread‑safe, unbounded priority‑blocking queue with elements ordered by priority. It requires elements to implement Comparable or a custom Comparator, blocks on take when empty, and does not accept null elements.

Thread safety : No external synchronization needed.

Unbounded : No capacity limit, so producers are not blocked by a full queue.

Priority ordering : Higher‑priority elements are dequeued first; default ordering uses Comparable, custom ordering via Comparator.

Blocking operations : take() blocks when empty; put() blocks only if the queue were bounded and full.

No null elements : Null values are rejected.

PriorityBlockingQueue can be used for priority‑based task scheduling and event handling.

Multi‑Priority Thread Pool Implementation

The author’s implementation follows three steps:

Create a task class that implements both java.lang.Comparable and java.lang.Runnable.

Define a priority field (int) to represent the task’s level.

Wrap the standard ThreadPoolExecutor with a PriorityBlockingQueue and provide helper methods to submit tasks with or without explicit priority.

Key code:

/**  Multi‑priority thread pool  */
static ThreadPoolExecutor levelPool = createPool(
    POOL_SIZE, POOL_SIZE, ALIVE_TIME,
    new PriorityBlockingQueue<PriorityTask>(),
    getFactory("L"),
    new ThreadPoolExecutor.DiscardPolicy());

static def executeLevel(PriorityTask task) {
    levelPool.execute(task);
}

static def executeLevel(int priority, Closure closure) {
    levelPool.execute(new PriorityTask(priority) {
        @Override
        void run() {
            closure();
        }
    });
}

static def executeLevel(Closure closure) {
    levelPool.execute(new PriorityTask(PRIORITY_LEVEL_DEFAULT) {
        @Override
        void run() {
            closure();
        }
    });
}

static abstract class PriorityTask implements Runnable, Comparable<PriorityTask> {
    int priority;
    PriorityTask(int priority) { this.priority = priority; }

    @Override
    public int compareTo(PriorityTask o) {
        return this.priority - o.priority;
    }
}

Test case submits ten tasks with decreasing priority values, prints the priority when each runs, and demonstrates that higher‑priority tasks execute first.

ThreadPoolUtil.getLevelPool().setCorePoolSize(1);
ThreadPoolUtil.getLevelPool().setMaximumPoolSize(1);
for (int i = 0; i < 10; i++) {
    ThreadPoolUtil.executeLevel(new ThreadPoolUtil.PriorityTask(10 - i) {
        @Override
        public void run() {
            SourceCode.sleep(0.01);
            System.out.println(this.getPriority());
        }
    });
}

Console output:

10
1
2
3
4
5
6
7
8
9

The result confirms that the priority‑based queue schedules tasks as expected.