ApsaraMQ for RocketMQ:メッセージフィルタリング

シナリオ

ApsaraMQ for RocketMQは、パブリッシュサブスクライブ (pub/sub) モデルに基づいたミドルウェアサービスで、アップストリームとダウンストリームビジネスの統合に使用されます。 実際のビジネスシナリオでは、トピック内のメッセージは、異なる消費ロジックを持つ異なるダウンストリームアプリケーションによって処理されます。 各アプリケーションは、その消費ロジックを満たすメッセージのみを消費する必要があります。

ApsaraMQ for RocketMQが提供するメッセージフィルタリング機能を使用して、コンシューマーが必要とするメッセージを効率的にフィルタリングできます。 これにより、多数の無効なメッセージがコンシューマに配信されるのを防ぎ、ダウンストリームシステムの作業負荷を軽減します。

ApsaraMQ for RocketMQのメッセージフィルタリング機能は、トピック内のメッセージをフィルタリングする際の問題点を解決し、同じ事業分野で詳細な分類が必要なシナリオで使用されます。 異なる事業分野のメッセージを管理する場合は、異なるトピックを使用することを推奨します。

概要

メッセージフィルタリングとは何ですか?

ApsaraMQ for RocketMQのメッセージフィルタリング機能は、コンシューマーが設定した条件に基づいてメッセージをフィルタリングし、条件を満たすメッセージをコンシューマーに送信します。

メッセージ属性とタグがプロデューサーとコンシューマーに定義されると、メッセージはApsaraMQ for RocketMQブローカーのフィルタリング条件に基づいてフィルタリングされ、照合されます。 フィルタリング条件を満たすメッセージのみがコンシューマに配信されます。

メッセージフィルタリングはどのように機能しますか?

分類

詳細については、「タグベースのフィルタリング」および「属性SQLベースのフィルタリング」をご参照ください。

サブスクリプションの一貫性

フィルタリング式は、サブスクリプションの一部を構成します。 ApsaraMQ for RocketMQのドメインモデルに基づいて、同じコンシューマーグループ内のコンシューマーのサブスクリプションのフィルタリング式は同じである必要があります。 そうしないと、特定のメッセージを消費できません。 詳細については、「サブスクリプション」をご参照ください。

タグベースのフィルタリング

タグベースのフィルタリングは、ApsaraMQ for RocketMQが提供する基本的なフィルタリング方法です。 このメソッドでは、プロデューサによって定義されたタグに基づいてメッセージが照合されます。 消費者はタグを使用して、消費するメッセージを指定します。

サンプルシナリオ

タグ設定

• プロデューサがメッセージを送信する前に、プロデューサは各メッセージに1つのタグのみを添付できます。

• 各タグは文字列で構成されています。 各タグの長さは128文字を超えないようにすることを推奨します。

• タグでは大文字と小文字が区別されます。 例えば、TAG Aとタグaは異なるタグです。

フィルタリングルール

サンプルコード

• メッセージ送信前にタグを指定する

  Message message = messageBuilder.setTopic("topic")
                  // The message key. You can use the key to search for the message. 
                  .setKeys("messageKey")
                  // The message tag. Consumers can use the tag to filter messages. 
                  // In this example, the tag of the message is set to TagA. 
                  .setTag("TagA")
                  // The message body. 
                  .setBody("messageBody".getBytes())
                  .build();
                      

• 単一のタグに一致するメッセージを購読する

  String topic = "Your Topic";
  // Subscribe to messages to which TagA is attached. 
  FilterExpression filterExpression = new FilterExpression("TagA", FilterExpressionType.TAG);
  pushConsumer.subscribe(topic, filterExpression);

• 複数のタグに一致するメッセージを購読する

  String topic = "Your Topic";
  // Subscribe to messages to which TagA, TagB, or TagC is attached. 
  FilterExpression filterExpression = new FilterExpression("TagA||TagB||TagC", FilterExpressionType.TAG);
  pushConsumer.subscribe(topic, filterExpression);

• トピック内のすべてのメッセージを購読する

  String topic = "Your Topic";
  // Subscribe to all messages. 
  FilterExpression filterExpression = new FilterExpression("*", FilterExpressionType.TAG);
  pushConsumer.subscribe(topic, filterExpression);

属性ベースのSQLフィルタリング

属性ベースのSQLフィルタリングは、ApsaraMQ for RocketMQが提供する高度なフィルタリング方法です。 このメソッドでは、メッセージはプロデューサによって指定されたキーと値のペアに基づいて照合されます。 プロデューサがメッセージを送信するとき、プロデューサは、各メッセージに対して複数の属性を指定できます。 次に、コンシューマは、特定のメッセージを受信するためにSQL式の属性を指定できます。

サンプルシナリオ

メッセージ属性の設定

プロデューサがメッセージを送信する前に、プロデューサは各メッセージのカスタム属性を指定することができる。 各属性は、カスタムのキーと値のペアです。

メッセージごとに複数の属性を指定できます。

フィルタリングルール

フィルター式を記述するときは、SQL-92の構文に従う必要があります。 次の表に、構文を示します。

サンプルコード

• メッセージ送信前にタグとカスタム属性を指定する

  Message message = messageBuilder.setTopic("topic")
                  // The message key. You can use the key to search for the message. 
                  .setKeys("messageKey")
                  // The message tag. Consumers can use the tag to filter messages. 
                  // In this example, the message tag is set to messageTag. 
                  .setTag("messageTag")
                  // You can also specify custom attributes for the messages, such as environment, region, and logical branch. 
                  // In this example, the custom attribute is region and the attribute value is Hangzhou. 
                  .addProperty("Region", "Hangzhou")
                  // The message body. 
                  .setBody("messageBody".getBytes())
                  .build();

• 単一の属性に一致するメッセージを購読する

  String topic = "topic";
  // Subscribe only to messages whose region attribute value is Hangzhou. 
  FilterExpression filterExpression = new FilterExpression("Region IS NOT NULL AND Region='Hangzhou'", FilterExpressionType.SQL92);
  simpleConsumer.subscribe(topic, filterExpression);

• 複数の属性に一致するメッセージを購読する

  String topic = "topic";
  // Subscribe to messages whose region attribute value is Hangzhou and price attribute value is greater than 30. 
  FilterExpression filterExpression = new FilterExpression("Region IS NOT NULL AND price IS NOT NULL AND Region = 'Hangzhou' AND price > 30", FilterExpressionType.SQL92);
  simpleConsumer.subscribe(topic, filterExpression);

• トピック内のすべてのメッセージを購読する

  String topic = "topic";
  // Subscribe to all messages. 
  FilterExpression filterExpression = new FilterExpression("True", FilterExpressionType.SQL92);
  simpleConsumer.subscribe(topic, filterExpression);

使用上の注意

トピックを適切に計画し、タグを指定する

よくある質問

複数のコンシューマーがトピック内の異なるタグをサブスクライブすると、メッセージが失われるのはなぜですか?

考えられる原因: コンシューマが同じコンシューマグループに属している場合、コンシューマがサブスクライブするタグは同じでなければなりません。 そうしないと、サブスクリプションの不整合が発生し、特定のメッセージが失われます。

フィルター条件を指定したときにコンシューマーが消費するメッセージの数を計算するにはどうすればよいですか。

コンシューマーが消費するメッセージの数は、フィルタリング条件を指定した後に計算されます。

オンライン消費者がメッセージを消費しなかったにもかかわらず、メッセージがグループに蓄積されるのはなぜですか?

SQLフィルタリングまたはタグベースのフィルタリングを使用してメッセージをフィルタリングすると、フィルタリング条件を満たさないメッセージが蓄積されます。 次の項目では、上記のシナリオでメッセージの蓄積数を計算する方法について説明します。

• SQLフィルタリング: 累積メッセージ数=準備完了メッセージ数 + 機内メッセージ数-フィルタリング条件を満たさないメッセージ数

• タグベースのフィルタリング: 累積メッセージ数=(準備完了メッセージ数 + 機内メッセージ数) × タグに一致するメッセージの割合

説明

タグに一致するメッセージの割合=サンプル内のタグに一致するメッセージの数 /サンプリングされたメッセージの総数

関連ドキュメント

メッセージングの完全なサンプルコードについては、「概要」をご参照ください。