Implementation of a Redis-Based Delay Queue in Java

Introduction – The author, a senior architect, shares a practical implementation of a delay queue inspired by a Youzan article, using Redis as the underlying storage.

Design Overview

The workflow consists of user task submission, pushing to a delay queue, storing metadata in a job pool, calculating execution time, placing a lightweight delay job in a bucket, and a timer component polling buckets to move ready tasks to a ready queue for consumption.

Key Components

Delay Queue (Redis sorted set) – handles message delivery.

Job Pool – stores full job metadata (ID, topic, delay, TTR, message, retry count, status).

Delay Bucket – holds DelayJob objects with execution timestamps.

Timer – scans buckets and triggers state transitions.

Ready Queue – per‑topic list where executable jobs are placed.

Task States

ready – executable.

delay – waiting for timer.

reserved – taken by consumer but not finished.

deleted – completed or removed.

Public API

Interface

Description

Data

add

Add task

Job data

pop

Retrieve pending task

Topic (task group)

finish

Mark task as finished

Task ID

delete

Delete task

Task ID

Implementation Details

Data Objects

Job entity:

@Data
@AllArgsConstructor
@NoArgsConstructor
public class Job implements Serializable {
    @JsonSerialize(using = ToStringSerializer.class)
    private Long id; // unique identifier
    private String topic; // business type
    private long delayTime; // delay duration
    private long ttrTime; // execution timeout
    private String message; // payload
    private int retryCount;
    private JobStatus status;
}

DelayJob wrapper:

@Data
@AllArgsConstructor
public class DelayJob implements Serializable {
    private long jodId;
    private long delayDate;
    private String topic;
    public DelayJob(Job job) {
        this.jodId = job.getId();
        this.delayDate = System.currentTimeMillis() + job.getDelayTime();
        this.topic = job.getTopic();
    }
    public DelayJob(Object value, Double score) {
        this.jodId = Long.parseLong(String.valueOf(value));
        this.delayDate = System.currentTimeMillis() + score.longValue();
    }
}

Containers

JobPool – simple K/V hash for job metadata:

@Component
@Slf4j
public class JobPool {
    @Autowired
    private RedisTemplate redisTemplate;
    private String NAME = "job.pool";
    private BoundHashOperations getPool() { return redisTemplate.boundHashOps(NAME); }
    public void addJob(Job job) { log.info("Add job: {}", JSON.toJSONString(job)); getPool().put(job.getId(), job); }
    public Job getJob(Long jobId) { Object o = getPool().get(jobId); return (o instanceof Job) ? (Job) o : null; }
    public void removeDelayJob(Long jobId) { log.info("Remove job: {}", jobId); getPool().delete(jobId); }
}

DelayBucket – Redis ZSet per bucket, round‑robin selection:

@Component
@Slf4j
public class DelayBucket {
    @Autowired private RedisTemplate redisTemplate;
    @Value("${thread.size}") private int bucketsSize;
    private List
bucketNames = new ArrayList<>();
    @Bean public List
createBuckets() { for (int i = 0; i < bucketsSize; i++) bucketNames.add("bucket" + i); return bucketNames; }
    private String getThisBucketName() { int idx = index.incrementAndGet() % bucketsSize; return bucketNames.get(idx); }
    private BoundZSetOperations getBucket(String name) { return redisTemplate.boundZSetOps(name); }
    public void addDelayJob(DelayJob job) { String name = getThisBucketName(); getBucket(name).add(job, job.getDelayDate()); log.info("Add delay job: {}", JSON.toJSONString(job)); }
    public DelayJob getFirstDelayTime(Integer index) { BoundZSetOperations bucket = getBucket(bucketNames.get(index)); Set
set = bucket.rangeWithScores(0, 1);
        if (CollectionUtils.isEmpty(set)) return null; ZSetOperations.TypedTuple tt = set.iterator().next(); Object v = tt.getValue(); return (v instanceof DelayJob) ? (DelayJob) v : null; }
    public void removeDelayTime(Integer index, DelayJob job) { getBucket(bucketNames.get(index)).remove(job); }
}

ReadyQueue – per‑topic list for executable jobs:

@Component
@Slf4j
public class ReadyQueue {
    @Autowired private RedisTemplate redisTemplate;
    private String NAME = "process.queue";
    private String getKey(String topic) { return NAME + topic; }
    private BoundListOperations getQueue(String topic) { return redisTemplate.boundListOps(getKey(topic)); }
    public void pushJob(DelayJob job) { log.info("Push to ready queue: {}", JSON.toJSONString(job)); getQueue(job.getTopic()).leftPush(job); }
    public DelayJob popJob(String topic) { Object o = getQueue(topic).leftPop(); if (o instanceof DelayJob) { log.info("Pop from ready queue: {}", JSON.toJSONString(o)); return (DelayJob) o; } return null; }
}

Timer Handling

A thread pool creates one handler per bucket to continuously move due jobs from DelayBucket to ReadyQueue and update their status.

@Component
public class DelayTimer implements ApplicationListener
{
    @Autowired private DelayBucket delayBucket;
    @Autowired private JobPool jobPool;
    @Autowired private ReadyQueue readyQueue;
    @Value("${thread.size}") private int length;
    @Override public void onApplicationEvent(ContextRefreshedEvent e) {
        ExecutorService es = new ThreadPoolExecutor(length, length, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());
        for (int i = 0; i < length; i++) {
            es.execute(new DelayJobHandler(delayBucket, jobPool, readyQueue, i));
        }
    }
}

Testing Endpoints

REST controller provides /delay/add , /delay/pop , /delay/finish , and /delay/delete to interact with the queue. Sample logs show tasks being added to the pool, placed in buckets, moved to the ready queue after the delay, and finally removed upon completion.

Conclusion

The article demonstrates a complete, runnable Redis delay‑queue solution in Java, suitable for distributed systems that require reliable delayed task execution, with clear component responsibilities, state management, and API contracts.