要件

開発プロセスにおいて、遅延キューを使用する必要があるシナリオが多数存在します。これは特殊なタイプのメッセージキューであり、メッセージをキューに送信しますが、即座にコンシューマに配信せず、一定時間後にメッセージを配信します。遅延キューの一般的な使用例を以下に示します：

• ECプラットフォームでの注文処理：注文後30分以内に支払いが行われない場合、自動的に注文をキャンセル
• 注文完了後、ユーザーが評価を行わない場合、5日後に自動的に高評価を付与
• 会員期限切れの15日前および3日前にSMS通知を送信
• 未払い状態の注文をタイムリーにクローズし、在庫を返却する方法
• 返金処理中の注文が正常に返金されたか定期的に確認する方法

実装アプローチ

遅延キューをサポートするミドルウェアは多数存在しますが、特に小規模なプロジェクトでは大規模なJMSミドルウェアを使用する必要がない場合があります。そのような場合、Redisを利用して遅延キューの機能を実現できます。

実装方法

分散メッセージコンポーネントeventbus-spring-boot-starterを使用して、Redisでの遅延メッセージと非同期即時メッセージの送受信を実装します。

バージョン要件

• SpringBoot 2.3.0.RELEASE以上
• Redis 5.0以上

クイックスタート

依存関係の追加

プロジェクトにeventbus-spring-boot-starterコンポーネントの依存関係を追加する必要があります：

<dependency>
    <groupId>com.github.likavn</groupId>
    <artifactId>eventbus-spring-boot-starter</artifactId>
    <version>2.5.1</version>
</dependency>

JSONシリアライゼーションツールとして、Fastjson2、Fastjson、Jackson、Gsonのいずれかをサポートしています。プロジェクトにspring-boot-starter-webが含まれている場合、Jacksonがデフォルトで含まれるため、追加のJSONツールは不要です。複数のJSONシリアライゼーションツールが存在する場合の優先順位は以下の通りです：

<!-- 任意のJSONシリアライゼーションツールを選択 -->
<dependency>
    <groupId>com.alibaba.fastjson2</groupId>
    <artifactId>fastjson2</artifactId>
    <version>${version}</version>
</dependency>

メッセージミドルウェアの設定

application.ymlファイルでメッセージエンジンのタイプを設定します：

eventbus:
  type: redis

Redis 5.0の新機能であるStreamを利用します。Redis Streamはメッセージの永続化とレプリケーション機能を提供し、クライアントが任意の時点のデータにアクセスできるようにし、各クライアントのアクセス位置を記憶し、メッセージの損失を防ぎます。デフォルトでは非ブロッキングポーリングを使用してStreamからメッセージを取得しますが、ブロッキングモードでの取得も設定可能です。

注：Redis 5.0～6.2未満のバージョンでは、期限切れメッセージの削除はStreamの長さの切り捨てによって実装されます。デフォルトではStream内のデータを10000件保持します。6.2以上のバージョンではメッセージのタイムアウト時間を設定でき、デフォルトで5日間のメッセージデータを保持します。

以下の依存関係を追加する必要があります：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-pool2</artifactId>
</dependency>

遅延メッセージの送受信

遅延メッセージの送信：

@Resource
private MessageDispatcher messageDispatcher;

// 非同期メッセージを送信
// 第一引数【DelayMessageHandler.class】は遅延メッセージの処理実装クラス
// 第二引数は遅延メッセージのボディオブジェクト
// 第三引数は遅延時間（秒単位）
messageDispatcher.sendDelayed(DelayMessageHandler.class, "order-12345", 10);

遅延メッセージリスナーの実装クラス：

/**
 * 遅延メッセージを購読
 * インターフェース【DelayedMessageListener】を実装し、コールバックメッセージのボディエンティティを設定
 */
@Slf4j
@Component
@MessageListener
public class DelayMessageHandler implements DelayedMessageListener<String> {
    @Override
    public void handle(Message<String> message) {
        String data = message.getBody();
        // メッセージ処理ロジック
    }
}

非同期メッセージの送受信

非同期メッセージの送信：

@Resource
private MessageDispatcher messageDispatcher;

// 非同期メッセージを送信
// 第一引数はビジネスメッセージコード
// 第二引数はビジネスメッセージのオブジェクトデータ
messageDispatcher.dispatch("order-processing", "payment-data");

非同期ビジネスメッセージリスナーの実装クラス：

/**
 * 非同期メッセージを購読
 * スーパークラス【MessageHandler】を継承し、監視するメッセージエンティティオブジェクトを設定
 */
@Slf4j
@Component
@MessageListener(codes="order-processing")
public class OrderMessageHandler implements MessageHandler<String> {

    // ビジネスメッセージボディオブジェクトデータを受信
    @Override
    public void handle(Message<String> message) {
        String payload = message.getBody();
        // メッセージ処理ロジック
    }
}

メッセージエンティティがMessageBodyインターフェースを実装し、エンティティ内でメッセージコードを定義することで、同じタイプのメッセージに対してリスナーの定義やメッセージ送信時に個別にメッセージコードを設定する必要がなくなります。

メッセージエンティティの定義：

// メッセージエンティティを定義
@Data
public class NotificationMessage implements MessageBody {
    private String text;
    
    @Override
    public String getCode() {
        return "NOTIFICATION_EVENT";
    }
}

NotificationMessageエンティティメッセージの送信：

// メッセージを送信
@Resource
private MessageDispatcher messageDispatcher;

// 非同期メッセージを送信
NotificationMessage notification = new NotificationMessage();
notification.setText("新着通知があります！");
// 第一引数はビジネスメッセージのオブジェクトデータ
messageDispatcher.dispatch(notification);

メッセージリスナーの定義（メッセージコードの指定不要）：

@Slf4j
@Component
@MessageListener
public class NotificationHandler implements MessageHandler<NotificationMessage> {

    @Override
    public void handle(Message<NotificationMessage> message) {
        NotificationMessage content = message.getBody();
        // 通知メッセージの処理ロジック
    }
}