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;
          }
      });                

サンプルコード

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

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

  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;
      }
  });