Understanding the Internal Working of Redisson DelayedQueue

The author investigates Redisson's DelayedQueue to clarify its internal execution flow, providing a step‑by‑step analysis that helps readers grasp the complete operation of a distributed delayed queue.

Basic Usage

Sending a delayed message (5 seconds) is demonstrated with the following code:

public void produce() {
  String queuename = "delay-queue";
  RBlockingQueue
blockingQueue = redissonClient.getBlockingQueue(queuename);
  RDelayedQueue
delayedQueue = redissonClient.getDelayedQueue(blockingQueue);
  delayedQueue.offer("测试延迟消息", 5, TimeUnit.SECONDS);
}

Receiving the message uses a blocking queue; the delayedQueue instance is created but not used directly:

public void consume() throws InterruptedException {
  String queuename = "delay-queue";
  RBlockingQueue
blockingQueue = redissonClient.getBlockingQueue(queuename);
  RDelayedQueue
delayedQueue = redissonClient.getDelayedQueue(blockingQueue);
  String msg = blockingQueue.take();
  // process the message
}

Both producer and consumer can run in separate Java processes as long as they connect to the same Redis instance.

Internal Data Structures

Redisson creates three queues: Message Delay Queue , Message Order Queue , and Message Target Queue . The delay queue stores messages ordered by expiration timestamp, while the order queue preserves insertion order (not used in the flow). When a message expires, it is moved from the first two queues into the target queue, from which consumers retrieve it.

Basic Flow

1. Send delayed message : the message is placed into the Delay Queue and Order Queue; if the Delay Queue was empty, a Pub/Sub notification is published. 2. Get delayed message : consumers block on the Target Queue and retrieve messages that have reached their expiration time. 3. Initialize delayed queue : a scheduler periodically checks the Delay Queue for the next expiration, moves due messages to the Target Queue, and sets a timer for the next check.

Sending Delayed Messages

The offer call eventually invokes RedissonDelayedQueue.offerAsync , which runs a Lua script that:

Inserts the message and its expiration timestamp into the Delay Queue (a sorted set) and the Order Queue (a list).

If the newly inserted message becomes the head of the Delay Queue, it publishes the expiration time on a dedicated channel so that all clients can schedule the next transfer.

@Override
public RFuture
offerAsync(V e, long delay, TimeUnit timeUnit) {
  if (delay < 0) {
    throw new IllegalArgumentException("Delay can't be negative");
  }
  long delayInMs = timeUnit.toMillis(delay);
  long timeout = System.currentTimeMillis() + delayInMs;
  long randomId = ThreadLocalRandom.current().nextLong();
  return commandExecutor.evalWriteNoRetryAsync(getRawName(), codec, RedisCommands.EVAL_VOID,
    "local value = struct.pack('dLc0', tonumber(ARGV[2]), string.len(ARGV[3]), ARGV[3]);"
    + "redis.call('zadd', KEYS[2], ARGV[1], value);"
    + "redis.call('rpush', KEYS[3], value);"
    + "local v = redis.call('zrange', KEYS[2], 0, 0); "
    + "if v[1] == value then "
    + "redis.call('publish', KEYS[4], ARGV[1]); "
    + "end;",
    Arrays.
asList(getRawName(), timeoutSetName, queueName, channelName),
    timeout, randomId, encode(e));
}

Getting Delayed Messages

Consumers simply call blockingQueue.take() , which internally performs a BLPOP on the Target Queue.

Initializing the Delayed Queue

Initialization creates a RedissonDelayedQueue instance and starts the QueueTransferTask :

public void init() {
    String queuename = "delay-queue";
    RBlockingQueue
blockingQueue = redissonClient.getBlockingQueue(queuename);
    RDelayedQueue
delayedQueue = redissonClient.getDelayedQueue(blockingQueue);
}

The task registers two Pub/Sub listeners on the channel redisson_delay_queue_channel:{queuename} :

On subscription, pushTask() is called, which eventually triggers pushTaskAsync() .

On receiving a new message, scheduleTask(startTime) is invoked to set a timer for the next transfer.

The core methods are:

scheduleTask(Long startTime) : calculates the delay until startTime and schedules a Netty Timeout that will call pushTask() when the time arrives.

pushTaskAsync() : runs a Lua script that moves up to 100 expired messages from the Delay Queue to the Target Queue and returns the next earliest expiration timestamp.

pushTask() : calls pushTaskAsync() , handles errors, and either reschedules itself immediately or after a short fallback interval.

These mechanisms ensure that even if the producer goes offline before its messages expire, any client that has initialized the delayed queue will eventually move the expired messages to the Target Queue, preventing data loss.

Summary

When the delayed queue is initialized, it processes any stale expired messages, determines the next expiration time, and sets up a timer or Pub/Sub listener to handle future messages. Sending a delayed message inserts it into both the Delay and Order queues and notifies listeners if it becomes the next due message. Consumers simply block on the Target Queue to retrieve ready messages.

Understanding these components helps avoid pitfalls such as lost messages when producers shut down prematurely.