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 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 you do not configure this attribute for a message or the attribute value of the message is not hangzhou, the message is not 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 only to messages that have tagA and whose value of the custom attribute price is 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 only to messages whose value of the custom attribute region is hangzhou and the 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);
  // To subscribe to all messages in the topic, set the value of the SQL expression to TRUE. 
  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 is not 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;
      }
  });