RabbitMQ Messaging & Async Systems · レッスン

Delayed Messagesプラグイン

RabbitMQ Delayed Message Exchangeプラグインを使用して、遅延メッセージ配信を実装します。カスタムタイマーを使わずに、将来の時点で処理するようメッセージをスケジュールします。

「Delayed Messagesプラグイン」はCoddyKit上の無料RabbitMQ Messaging & Async Systemsレッスンです。これはレッスン1/4です。下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。これはRabbitMQ Messaging & Async Systems学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 RabbitMQ Messaging & Async Systemsコースには全4レッスンが含まれています。

このレッスンの一部はまだ翻訳されておらず、英語で表示されています。

What are Delayed Messages?

Sometimes you don't want a message processed immediately. Think of sending a reminder email an hour from now, or processing a payment 30 minutes after an order is placed.

Delayed messages allow you to publish a message to RabbitMQ, but tell the broker to hold onto it for a specific duration before delivering it to consumers.

Scheduling Without the Plugin

Without a dedicated feature, scheduling messages can be complex:

Custom Timers: You might build your own service with timers, but this adds complexity and a single point of failure.
Polling Databases: Storing messages in a database and periodically checking for due times is inefficient.
External Schedulers: Using CRON jobs or other external schedulers still requires your application to manage the message state.

RabbitMQ's Delayed Message Plugin simplifies this greatly!

Meet the `x-delayed-message` Plugin

The RabbitMQ Delayed Message Exchange plugin provides a special exchange type that can hold messages and release them after a specified delay.

It acts like a buffer, managing the delay internally without requiring your application to keep track of message timings.

This makes scheduling messages much simpler and more robust.

Delayed Exchange Mechanics

When you publish a message to an `x-delayed-message` exchange, you include a special header: x-delay. This header's value is the delay in milliseconds.

The exchange holds the message until its delay expires. Once expired, the message is routed to queues exactly as if it were a regular message published to an exchange of its original type (e.g., direct, fanout, topic).

Enabling the Plugin (Admin)

Before using it, the plugin must be enabled on your RabbitMQ server. This is typically done via the command line:

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

A server restart might be required for the changes to take effect. Always ensure your broker has the plugin enabled before attempting to use it.

Declaring a Delayed Exchange

To use delayed messages, you first declare an exchange with the type x-delayed-message. You also specify its underlying type (e.g., direct, fanout, topic) which determines how messages are routed after the delay.

Here's how to declare a delayed direct exchange in Java:

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.util.HashMap;
import java.util.Map;

public class DeclareDelayedExchange {
    private static final String EXCHANGE_NAME = "my_delayed_exchange";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost"); // Assuming RabbitMQ is local
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {

            Map<String, Object> args = new HashMap<>();
            args.put("x-delayed-type", "direct"); // Underlying exchange type

            channel.exchangeDeclare(EXCHANGE_NAME, "x-delayed-message", true, false, args);
            System.out.println("Delayed exchange '" + EXCHANGE_NAME + "' declared.");
        }
    }
}

Sending with `x-delay` Header

When publishing to your x-delayed-message exchange, you add an x-delay header to your message properties. The value is an integer representing the delay in milliseconds.

Let's send a message that will be delivered after 5 seconds (5000 milliseconds).

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.util.HashMap;
import java.util.Map;

public class DelayedMessageProducer {
    private static final String EXCHANGE_NAME = "my_delayed_exchange";
    private static final String ROUTING_KEY = "delayed.key";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactoryFactory();
        factory.setHost("localhost");
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {

            // Ensure the delayed exchange is declared (from previous scene)
            Map<String, Object> args = new HashMap<>();
            args.put("x-delayed-type", "direct");
            channel.exchangeDeclare(EXCHANGE_NAME, "x-delayed-message", true, false, args);
            System.out.println("Exchange declared (if not exists).");

            String message = "Hello, delayed world!";
            int delayInMs = 5000; // 5 seconds

            Map<String, Object> headers = new HashMap<>();
            headers.put("x-delay", delayInMs); // Set the delay header

            AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                .headers(headers)
                .build();

            channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, props, message.getBytes("UTF-8"));
            System.out.println(" [x] Sent '" + message + "' with delay " + delayInMs + "ms");
        }
    }
}

Consuming Delayed Messages

Consumers don't need any special logic to receive delayed messages. Once the delay expires, the x-delayed-message exchange routes the message to the bound queues, and consumers receive it like any other message.

The key is that the message arrives at the consumer after the specified delay. You'll need to bind a queue to your delayed exchange with the appropriate routing key.

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import java.nio.charset.StandardCharsets;
import java.util.Date;

public class DelayedMessageConsumer {
    private static final String EXCHANGE_NAME = "my_delayed_exchange";
    private static final String QUEUE_NAME = "delayed_queue";
    private static final String ROUTING_KEY = "delayed.key";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        // Declare the queue and bind it to the delayed exchange
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
            System.out.println(" [x] Received '" + message + "' at " + new Date());
        };
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {});
    }
}

Common Use Cases

Delayed messages are incredibly useful for:

Reminders: Send a notification after a user's trial expires.
Scheduled Tasks: Process a batch job at a specific time in the future.
Retry Mechanisms: Requeue a failed message to be retried after a delay.
Drip Campaigns: Send a series of emails over several days.

They remove the need for complex external scheduling services.

Quick Check: Delayed Message Basics

You've learned how to declare an x-delayed-message exchange and publish messages with a delay. Let's test your understanding.

Recap: Delayed Messages Plugin

In this lesson, you learned about the RabbitMQ Delayed Message Exchange plugin. It allows you to schedule messages to be delivered at a future time.

You declare an exchange with type x-delayed-message and an underlying exchange type.
You publish messages with an x-delay header (value in milliseconds).
Consumers receive these messages normally, but only after the specified delay.

This powerful feature simplifies many time-based messaging patterns.

よくある質問

「Delayed Messagesプラグイン」レッスンは無料ですか？

はい。「Delayed Messagesプラグイン」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、RabbitMQ Messaging & Async Systemsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 RabbitMQ Messaging & Async Systemsコースには全4レッスンが含まれています。

「Delayed Messagesプラグイン」で何を学びますか？

RabbitMQ Delayed Message Exchangeプラグインを使用して、遅延メッセージ配信を実装します。カスタムタイマーを使わずに、将来の時点で処理するようメッセージをスケジュールします。ブラウザで直接実行するハンズオンコードでRabbitMQ Messaging & Async Systemsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。

RabbitMQ Messaging & Async Systemsを始めるのに経験は必要ですか？

事前経験は必要ありません。CoddyKitのRabbitMQ Messaging & Async Systemsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。これはレッスン1/4です。

「Delayed Messagesプラグイン」レッスンにはどのくらい時間がかかりますか？

ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。

このRabbitMQ Messaging & Async Systemsレッスンでコードを書いて実行できますか？

はい。すべてのRabbitMQ Messaging & Async Systemsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。

このコースのすべてのレッスン

Delayed Messagesプラグイン
フェデレーション向けShovelプラグイン
クラスタリング向けFederationプラグイン
メッセージの重複排除とConsistent Hash Exchangeプラグイン

← RabbitMQ Messaging & Async Systemsに戻る