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

説明

メッセージフィルタリング機能は、メッセージ属性とフィルタ条件を設定することによって実装されます。 メッセージ属性を設定して、プロデューサーがトピックに送信するメッセージを分類できます。 さらに、コンシューマーがトピックで指定された属性を持つメッセージをサブスクライブできるように、フィルター条件を設定する必要があります。 このようにして、ブローカは、プロデューサから送信されたメッセージをフィルタリングし、指定されたフィルタ条件を満たすメッセージのみをコンシューマに配信することができる。

コンシューマーがトピックをサブスクライブしているが、フィルター条件が指定されていない場合、プロデューサーから送信されたメッセージに属性が設定されているかどうかに関係なく、トピック内のすべてのメッセージがコンシューマーに配信されます。

タグを使用したメッセージのフィルタリング

タグは、トピック内のメッセージを分類するラベルです。 ApsaraMQ for RocketMQプロデューサーがメッセージを送信するときに、メッセージにタグを追加できます。 消費者は、メッセージに添付されたタグに基づいてメッセージを購読できます。

シナリオ

設定方法

ApsaraMQ for RocketMQでは、タグを使用してメッセージをフィルタリングするためのコードをクライアントSDKで定義できます。 プロデューサーがメッセージを送信するときにメッセージにタグを追加し、コンシューマーがサブスクライブするメッセージのタグを指定する必要があります。 クライアントSDKの詳細については、「」をご参照ください。 次の情報は、プロデューサーとコンシューマーのコードを定義する方法を示しています。

• メッセージを送信する

  メッセージを送信する前に、各メッセージのタグを指定します。

      Message msg = new Message("MQ_TOPIC","TagA","Hello MQ".getBytes());                

• すべてのメッセージを購読する

  トピック内のすべてのメッセージをサブスクライブするようにコンシューマーを構成する場合は、アスタリスク (*) を使用してすべてのタグを示します。

  consumer.subscribe("MQ_TOPIC", "*", new MessageListener() {
          public Action consume(Message message, ConsumeContext context) {
              System.out.println(message.getMsgID());
              return Action.CommitMessage;
          }
      });                

• 特定のタイプのメッセージを購読する

  トピック内の特定のタイプのメッセージをサブスクライブするようにコンシューマーを構成する場合は、対応するタグを指定します。

  consumer.subscribe("MQ_TOPIC", "TagA", new MessageListener() {
          public Action consume(Message message, ConsumeContext context) {
              System.out.println(message.getMsgID());
              return Action.CommitMessage;
          }
      });                

• 複数のタイプのメッセージを購読する

  トピック内の複数のタイプのメッセージをサブスクライブするようにコンシューマーを構成する場合は、対応するタグを指定し、2つの縦棒 (| |) で区切ります。

  consumer.subscribe("MQ_TOPIC", "TagA | | TagB", new MessageListener() {
          public Action consume(Message message, ConsumeContext context) {
              System.out.println(message.getMsgID());
              return Action.CommitMessage;
          }
      });                

• エラーの例

  コンシューマーにトピックへの複数のサブスクリプションがあり、各サブスクリプションに異なるタグがある場合、コンシューマーは、タグが最新のサブスクリプションで指定されたタグと一致するメッセージのみを受信します。

      // In the following code, a consumer can receive only messages with TagB in MQ_TOPIC but cannot receive messages with TagA. 
      consumer.subscribe("MQ_TOPIC", "TagA", new MessageListener() {
          public Action consume(Message message, ConsumeContext context) {
              System.out.println(message.getMsgID());
              return Action.CommitMessage;
          }
      });
      consumer.subscribe("MQ_TOPIC", "TagB", new MessageListener() {
          public Action consume(Message message, ConsumeContext context) {
              System.out.println(message.getMsgID());
              return Action.CommitMessage;
          }
      });                

SQL式を使用したメッセージのフィルタリング

SQL式を使用してメッセージをフィルターするには、次の手順を実行します。プロデューサーがメッセージを送信するときにメッセージのカスタム属性を設定し、SQL構文を使用してフィルター式を定義し、コンシューマーがサブスクライブするメッセージの属性を指定します。 ApsaraMQ for RocketMQは、カスタム属性がフィルター式の計算結果と一致するメッセージを選択し、これらのメッセージをコンシューマーに配信します。

制限事項

シナリオ

設定方法

サンプルコード

• メッセージを送信します。

  メッセージのタグとカスタム属性を設定します。

  Producer producer = ONSFactory.createProducer(properties);
  // Set the value of Tag to tagA. 
  Message msg = new Message("topicA", "tagA", "Hello MQ".getBytes());
  // Set the value of the custom attribute region to hangzhou. 
  msg.putUserProperties("region", "hangzhou");
  // Set the value of the custom attribute price to 50. 
  msg.putUserProperties("price", "50");
  SendResult sendResult = producer.send(msg);

• カスタム属性に基づいてメッセージをサブスクライブします。

  Consumer consumer = ONSFactory.createConsumer(properties);
  // Subscribe to only messages whose value of the custom attribute region is hangzhou. If this attribute is not configured for a message or the attribute value of the message is not hangzhou, the message will not be delivered to the consumer. 
  consumer.subscribe("topicA", MessageSelector.bySql("region IS NOT NULL AND region = 'hangzhou'"), new MessageListener() {
      public Action consume(Message message, ConsumeContext context) {
          System.out.printf("Receive New Messages: %s %n", message);
          return Action.CommitMessage;
      }
  });

  期待される結果: この例で送信されるメッセージにはカスタム属性領域があり、属性値はhangzhouです。 メッセージはフィルタ条件を満たし、消費者に配信されます。

• タグとカスタム属性の両方に基づいてメッセージをサブスクライブします。

  Consumer consumer = ONSFactory.createConsumer(properties);
  // Subscribe to only messages with tagA and their values of the custom attribute price are greater than 30. 
  consumer.subscribe("topicA", MessageSelector.bySql("TAGS IS NOT NULL AND price IS NOT NULL AND TAGS = 'tagA' AND price > 30 "), new MessageListener() {
      public Action consume(Message message, ConsumeContext context) {
          System.out.printf("Receive New Messages: %s %n", message);
          return Action.CommitMessage;
      }
  });

  期待される結果: この例で送信されるメッセージにはtagAとカスタム属性価格の両方があり、カスタム属性価格の値は30を超えています。 メッセージはフィルタ条件を満たし、消費者に配信されます。

• 複数のカスタム属性に基づいてメッセージを購読します。

  Consumer consumer = ONSFactory.createConsumer(properties);
  // Subscribe to only messages whose value of the custom attribute region is hangzhou and value of the custom attribute price is less than 20. 
  consumer.subscribe("topicA", MessageSelector.bySql("region IS NOT NULL AND price IS NOT NULL AND region = 'hangzhou' AND price < 20"), new MessageListener() {
      public Action consume(Message message, ConsumeContext context) {
          System.out.printf("Receive New Messages: %s %n", message);
          return Action.CommitMessage;
      }
  });

  期待される結果: メッセージがフィルター条件を満たしていないため、メッセージはコンシューマーに配信されません。 消費者は、カスタム属性価格の値が20未満のメッセージを購読しますが、作成者のメッセージに設定されたカスタム属性価格の値は50です。

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

  Consumer consumer = ONSFactory.createConsumer(properties);
  // Set the value of the SQL expression to TRUE to subscribe to all messages in the topic. 
  consumer.subscribe("topicA", MessageSelector.bySql("TRUE"), new MessageListener() {
      public Action consume(Message message, ConsumeContext context) {
          System.out.printf("Receive New Messages: %s %n", message);
          return Action.CommitMessage;
      }
  });

  期待される結果: トピック内のすべてのメッセージがコンシューマーに配信されます。

• エラーの例

  カスタム属性は、プロデューサーがメッセージを送信するときにメッセージに対して構成されておらず、カスタム属性が存在するかどうかをチェックするためのロジックはフィルター式に定義されていません。 カスタム属性は、式のフィルター条件として直接使用されます。 この場合、メッセージは消費者に配信されません。

  Consumer consumer = ONSFactory.createConsumer(properties);
  // The message attribute product is not configured during message sending. The filtering fails and the message will not be delivered to the consumer. 
  consumer.subscribe("topicA", MessageSelector.bySql("product = 'MQ'"), new MessageListener() {
      public Action consume(Message message, ConsumeContext context) {
          System.out.printf("Receive New Messages: %s %n", message);
          return Action.CommitMessage;
      }
  });               

関連ドキュメント

• 同じグループIDを使用するコンシューマインスタンスは、同じトピックをサブスクライブする必要があります。 詳細については、「サブスクリプションの一貫性」をご参照ください。

• トピックとタグを使用して、さまざまなサービスのメッセージを分類できます。 詳細については、「トピックとタグのベストプラクティス」をご参照ください。