アノテーション駆動型のリスナーエンドポイント

メッセージを非同期で受信する最も簡単な方法は、アノテーション付きリスナーエンドポイントインフラストラクチャを使用することです。簡単に言うと、マネージド Bean のメソッドを Redis リスナーエンドポイントとして公開できます。次の例は、その使用方法を示しています。

@Component
public class MyService {

  @RedisListener(topic = "my-channel")
  public void processOrder(String data) { ... }
}

前述の例の考え方は、チャネル my-channel を介してメッセージが受信されるたびに、それに応じて processOrder メソッドが呼び出されるということです (この場合、MessageListenerAdapter が提供するものと同様に、Redis Pub/Sub メッセージの内容が使用されます)。

アノテーション付きエンドポイントインフラストラクチャは、アノテーション付きメソッドごとにバックグラウンドでメッセージリスナーを作成し、RedisMessageListenerContainer に登録します。エンドポイントはアプリケーションコンテキストには登録されませんが、RedisListenerEndpointRegistry Bean を使用することで管理目的で簡単に見つけることができます。

@RedisListener は繰り返し可能なアノテーションなので、@RedisListener 宣言を追加することで、複数の Redis トピック（チャネルまたはパターン）を同じメソッドに関連付けることができます。

リスナーエンドポイントアノテーションを有効にする

@RedisListener アノテーションのサポートを有効にするには、次の例に示すように、@EnableRedisListeners を @Configuration クラスの 1 つに追加できます。

Java
Kotlin

@Configuration
@EnableRedisListeners
public class RedisConfiguration {

  @Bean
  public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory connectionFactory) {

    RedisMessageListenerContainer factory = new RedisMessageListenerContainer();
    factory.setConnectionFactory(connectionFactory);
    return factory;
  }
}

@Configuration
@EnableRedisListeners
class RedisConfiguration {

  @Bean
  fun redisMessageListenerContainer(connectionFactory: RedisConnectionFactory) =
    RedisMessageListenerContainer().apply {
      setConnectionFactory(connectionFactory);
    }
}

RedisListenerConfigurer インターフェースを実装することで、リスナーレジストラ (Javadoc) をカスタマイズできます。詳細と例については、RedisListenerConfigurer (Javadoc) を実装するクラスの Javadoc を参照してください。

プログラムによるエンドポイント登録

RedisListenerEndpoint は Redis エンドポイントのモデルを提供し、そのモデルに合わせてコンテナーを構成するロールを担います。このインフラストラクチャでは、RedisListener アノテーションによって検出されたエンドポイントに加えて、プログラムによってエンドポイントを構成できます。以下の例は、その方法を示しています。

@Configuration
@EnableRedisListeners
public class AppConfig implements RedisListenerConfigurer {

  @Override
  public void configureRedisListeners(RedisListenerEndpointRegistrar registrar) {
    SimpleRedisListenerEndpoint endpoint = new SimpleRedisListenerEndpoint();
    endpoint.setId("myRedisEndpoint");
    endpoint.setTopic("my-channel");
    endpoint.setMessageListener((message, pattern) -> {
      // processing
    });
    registrar.registerEndpoint(endpoint);
  }
}

前の例では、実際の MessageListener を呼び出して提供する SimpleRedisListenerEndpoint を使用しました。ただし、独自のエンドポイントバリアントを作成して、カスタム呼び出しメカニズムを記述することもできます。

@RedisListener の使用を完全にスキップし、RedisListenerConfigurer を介してエンドポイントのみをプログラムで登録できることに注意してください。

アノテーション付きエンドポイントメソッドシグネチャー

これまでエンドポイントに単純な String を注入してきましたが、実際には非常に柔軟なメソッドシグネチャーを持つことができます。次の例では、ヘッダー付きの Order を注入するように書き換えています。

@Component
public class MyService {

  @RedisListener("my-order-channels*")
  public void processOrder(Order order, @Header("pattern") Topic pattern) {
    ...
  }
}

Redis リスナーエンドポイントに注入できる主な要素は以下のとおりです。

受信した Redis メッセージを表す org.springframework.messaging.Message。このメッセージには、ヘッダー（PubSubHeaders で定義）が含まれていることに注意してください。
@Header - 特定のヘッダー値を抽出するためのアノテーション付きメソッド引数。Redis Pub/Sub メッセージは本文のみで構成されているため、メッセージに一致したトピックやパターンなどのヘッダー値は、合成的にヘッダーとして追加されます。
すべてのヘッダーにアクセスするために java.util.Map にも割り当て可能である必要がある @Headers アノテーション付き引数。
サポートされている型以外の、アノテーションのない要素はペイロードとみなされます。@Payload でパラメーターにアノテーションを付けることで、これを明示的に指定できます。また、@Valid を追加することで検証を有効にすることもできます。

Spring の Message 抽象化を挿入する機能は、トランスポート固有の API に依存することなく、トランスポート固有のメッセージに格納されているすべての情報を活用するのに特に役立ちます。次の例は、その方法を示しています。

@RedisListener("my-channel")
public void processOrder(Message<byte[]> order) { ... }

メソッド引数の処理は DefaultMessageHandlerMethodFactory によって提供され、追加のメソッド引数をサポートするようにさらにカスタマイズできます。アノテーションベースのエンドポイントは柔軟なメッセージ変換をサポートしており、これは RedisMessageConverters によって提供され、String、バイト配列、JSON 用のデフォルトのコンバーターセット (サポートされているライブラリがクラスパスに存在する場合) も含まれています。メッセージ変換をカスタマイズする場合は、次の例に示すように、RedisListenerConfigurer を実装して configureMessageConverters メソッドをオーバーライドすることでカスタマイズできます。

@Configuration
@EnableRedisListeners
public class AppConfig implements RedisListenerConfigurer {

  @Override
  public void configureMessageConverters(RedisMessageConverters.Builder builder) {
    builder.withStringConverter(StandardCharsets.UTF_8)
      .addCustomConverter(new JdkSerializationRedisSerializer());
  }
}

RedisMessageConverters (Javadoc) はデフォルトで以下のコンバーターを登録します。

StringMessageConverter
ByteArrayMessageConverter
JSON コンバーター (クラスパス上に Jackson、Gson、JSON-B、Kotlin Serialization などのサポート対象ライブラリが存在する場合)

RedisMessageConverters は、JSON 直列化に Spring Data Redis の RedisSerializers を使用します。その JSON 直列化フォーマットと動作は、Spring Messaging の JacksonJsonMessageConverter とは若干異なる場合があります。Spring Messaging のバリアントを使用する場合は、Builder.addCustomConverter(MessageConverter) を通じて目的のコンバーターを設定してください。

@RedisListener(consumes) は、異なるメッセージコンバーターを使用する際に、適切なコンバーターを選択するために、目的のコンテンツ型を指定できます。次の例は、前の例と同様の登録を前提とした、JdkSerializationRedisSerializer コンバーターのコンバーター選択を示しています。

@RedisListener(topic = "my-channel", consumes = JdkSerializerMessageConverter.APPLICATION_JAVA_SERIALIZED_OBJECT_VALUE)
public void processOrder(Person person) { ... }

コンテンツ型を指定しないアノテーション付きエンドポイントのペイロード引数変換は、メッセージペイロードを読み取ることができる最初のコンバーターによって処理されます。

メソッド引数の解決をより細かく制御したい場合は、MessageHandlerMethodFactory から RedisListenerConfigurer までのカスタム MessageHandlerMethodFactory を設定することもできます。そこで変換と検証のサポートをカスタマイズすることも可能です。

たとえば、処理する前に Order が有効であることを確認したい場合、次の例に示すように、ペイロードに @Valid でアノテーションを付け、必要なバリデーターを構成できます。

@Configuration
@EnableRedisListeners
public class AppConfig implements RedisListenerConfigurer {

  @Override
  public void configureRedisListeners(RedisListenerEndpointRegistrar registrar) {
    registrar.setMessageHandlerMethodFactory(myRedisHandlerMethodFactory());
  }

  @Bean
  public DefaultMessageHandlerMethodFactory myRedisHandlerMethodFactory() {
    DefaultMessageHandlerMethodFactory factory = new DefaultMessageHandlerMethodFactory();
    factory.setValidator(myValidator());
    return factory;
  }
}