FAQ: How do I use the delayed message exchange in LavinMQ?

Learn how to schedule messages for future delivery with LavinMQ's delayed message exchange. This FAQ explains what it is, how it works internally, and how to set it up and use it.

What is a delayed message exchange?

A delayed message exchange in LavinMQ holds messages for a specified period before routing them to their destination. The delay is set using the x-delay header in each message. This feature is valuable when you need to control the timing of message delivery. Everyday use cases include:

• Scheduled tasks: Execute operations at predetermined times.
• Retries: Delay the reprocessing of messages following an error.
• Time-sensitive workflows: Coordinate processes that depend on precise timing.

Declaring a delayed exchange in LavinMQ

You can set up a delayed exchange through code (recommended) or by using the LavinMQ management interface.

Declaring a delayed exchange in LavinMQ requires specifying the x-delayed-type argument. This argument indicates the underlying exchange type ( e.g., direct , fanout , topic , or headers ), providing flexibility in how messages are routed after the delay.

Example in Ruby:

require "amqp-client"
require "amq-protocol"

# Connect to LavinMQ
connection = AMQP::Client::Connection.new("amqp://localhost")
channel = connection.channel

# Define the exchange details
exchange_name = "delayed_fanout_exchange"
exchange_type = "x-delayed-message"
arguments = {
  "x-delayed-type" => "fanout"  # Specify the underlying exchange type
}

# Declare the delayed exchange
channel.exchange_declare(
  exchange: exchange_name,
  type: exchange_type,
  durable: true,
  arguments: arguments
)

puts "Delayed exchange '#exchange_name' declared successfully."

# Close the channel and connection
channel.close
connection.close

How to publish a delayed message

When publishing, include the desired delay in the x-delay header to send a delayed message. This header tells LavinMQ how long it should hold the message before delivering it to the destination queue.

Example in Ruby:

require "amqp-client"
require "amq-protocol"

# Connect to LavinMQ
connection = AMQP::Client::Connection.new("amqp://localhost")
channel = connection.channel

message = "This is a delayed message for LavinMQ."
headers = {
  "x-delay" => 2500_i64  # Delay in milliseconds, required as Int64
}

# Publish the message to the delayed exchange
channel.basic_publish(
  exchange: exchange_name,
  routing_key: "", # Fanout exchanges ignore the routing key
  body: message,
  properties: AMQP::Client::BasicProperties.new(headers: headers)
)

puts "Message published to LavinMQ with a 2,500ms delay."

# Close the channel and connection
channel.close
connection.close

How does LavinMQ handle delayed messages behind the scenes?

When you declare a delayed exchange, LavinMQ automatically creates an internal queue to store delayed messages. This internal queue is pre-configured with a dead-letter-exchange argument, which handles messages once their delay period expires. Here’s the process:

1. Message reception: A message published to the delayed exchange is routed to an internal queue.
2. Delay storage: The message is stored with a delay defined by its x-delay header , and the internal queue sets a corresponding Time-To-Live (TTL).
3. Expiration and republish: Once the TTL expires, the message is removed from the internal queue and republished to a dead-letter exchange.
4. Final routing: The republished message, now without a delay, is routed directly to the destination queue.

A special message store, **DelayedMessageStore**, manages this internal queue. Rather than using standard FIFO ordering, it organizes messages based on their delay values, ensuring that messages with shorter delays expire first.

Conclusion

LavinMQ's delayed message exchange offers an efficient solution for managing scheduled message delivery. By declaring a delayed exchange with the x-delayed-type argument and publishing messages with an x-delay header, messages can be reliably delayed and routed as needed. This feature is ideal for implementing scheduled tasks, handling retries, and managing other time-based workflows.

For more detailed information or advanced configurations, refer to the LavinMQ documentation or seek support from our engineers.