package pqueue
type Job struct {
ID string
Payload interface{}
Priority int
seq uint64
}
type jobHeap []*Job
func (h jobHeap) Len() int { return len(h) }
func (h jobHeap) Less(i, j int) bool {
if h[i].Priority != h[j].Priority {
return h[i].Priority > h[j].Priority // higher priority first
}
return h[i].seq < h[j].seq // FIFO among equal priorities
}
func (h jobHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] }
func (h *jobHeap) Push(x interface{}) {
*h = append(*h, x.(*Job))
}
func (h *jobHeap) Pop() interface{} {
old := *h
n := len(old)
item := old[n-1]
old[n-1] = nil
*h = old[:n-1]
return item
}package pqueue
import (
"container/heap"
"sync"
)
type PriorityQueue struct {
mu sync.Mutex
cond *sync.Cond
h jobHeap
seq uint64
closed bool
}
func New() *PriorityQueue {
pq := &PriorityQueue{h: make(jobHeap, 0, 16)}
pq.cond = sync.NewCond(&pq.mu)
heap.Init(&pq.h)
return pq
}
func (pq *PriorityQueue) Enqueue(j *Job) bool {
pq.mu.Lock()
defer pq.mu.Unlock()
if pq.closed {
return false
}
pq.seq++
j.seq = pq.seq
heap.Push(&pq.h, j)
pq.cond.Signal()
return true
}
func (pq *PriorityQueue) Dequeue() (*Job, bool) {
pq.mu.Lock()
defer pq.mu.Unlock()
for pq.h.Len() == 0 && !pq.closed {
pq.cond.Wait()
}
if pq.h.Len() == 0 {
return nil, false
}
j := heap.Pop(&pq.h).(*Job)
return j, true
}
func (pq *PriorityQueue) Close() {
pq.mu.Lock()
pq.closed = true
pq.mu.Unlock()
pq.cond.Broadcast()
}package pqueue
import (
"log"
"sync"
)
type Handler func(*Job)
type Pool struct {
pq *PriorityQueue
wg sync.WaitGroup
}
func NewPool(pq *PriorityQueue, workers int, h Handler) *Pool {
p := &Pool{pq: pq}
for i := 0; i < workers; i++ {
p.wg.Add(1)
go p.run(i, h)
}
return p
}
func (p *Pool) run(id int, h Handler) {
defer p.wg.Done()
for {
job, ok := p.pq.Dequeue()
if !ok {
log.Printf("worker %d: queue drained, exiting", id)
return
}
h(job)
}
}
func (p *Pool) Shutdown() {
p.pq.Close()
p.wg.Wait()
}This snippet builds a concurrency-safe priority job queue in Go using the standard library's container/heap. The core idea is that jobs are not processed first-in-first-out but by an explicit priority, with ties broken by insertion order so equal-priority jobs stay stable and never starve one another.
The pqueue.go tab implements the raw heap. container/heap does not provide a container; it provides algorithms that operate on any type satisfying heap.Interface, which extends sort.Interface with Push and Pop. The jobHeap slice implements Len, Less, Swap, Push, and Pop. A crucial detail is that Less compares priority first and falls back to seq for equal priorities, giving deterministic FIFO ordering among equal jobs. Another subtlety Go beginners miss: Push and Pop on heap.Interface operate on the slice's tail and are called by the package's heap.Push/heap.Pop helpers — application code must never manipulate the slice directly, or the heap invariant breaks.
The PriorityQueue type in queue.go wraps jobHeap with a sync.Mutex and a sync.Cond. The condition variable is the key to blocking consumers cleanly: Dequeue loops on cond.Wait() until either a job is available or the queue is closed, avoiding busy-waiting. Enqueue assigns a monotonically increasing seq under the lock so tie-breaking is consistent, then calls cond.Signal() to wake one waiter. Close sets a flag and calls cond.Broadcast() so every parked goroutine wakes and returns, which is essential for a clean shutdown without leaked goroutines.
The worker.go tab shows the queue in action with a Pool of goroutines. Each worker calls Dequeue in a loop; when the queue is closed and drained, Dequeue returns ok == false and the worker exits. pool.Shutdown closes the queue and waits on a sync.WaitGroup so all in-flight work finishes.
This pattern is the right tool when work items have differing urgency — think of retry jobs, user-facing requests versus batch jobs, or deadline-driven scheduling. The trade-off is O(log n) enqueue/dequeue rather than O(1), and the mutex serializes access, so extremely high-throughput systems may prefer sharded queues. For most job-processing workloads, this compact, correct implementation is more than sufficient.
Related snips
package api
import (
"net/http"
"runtime/debug"
)Expose build metadata for debugging deploys
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
Share this code
Here's the card — post it anywhere.
