ApsaraMQ for RocketMQ:Penyaringan Pesan

Skema Contoh

Metode Konfigurasi

ApsaraMQ for RocketMQ memungkinkan Anda mendefinisikan kode dalam SDK klien untuk memfilter pesan menggunakan tag. Sebelum produsen mengirimkan pesan, Anda harus menambahkan tag ke pesan dan menentukan tag dari pesan yang ingin diterima oleh konsumen. Untuk informasi tentang SDK, lihat Ikhtisar. Bagian berikut menjelaskan cara mendefinisikan kode untuk produsen dan konsumen:

• Mengirim pesan

  Sebelum Anda mengirim pesan, tentukan tag untuk setiap pesan.

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

• Berlangganan semua pesan

  Jika konsumen ingin berlangganan semua pesan dalam topik, gunakan tanda bintang (*) untuk menentukan semua tag.

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

• Berlangganan jenis pesan tertentu

  Jika konsumen ingin berlangganan jenis pesan tertentu dalam topik, tentukan tag yang sesuai.

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

• Berlangganan beberapa jenis pesan

  Jika konsumen ingin berlangganan beberapa jenis pesan dalam topik, tentukan tag yang sesuai. Pisahkan beberapa tag dengan dua garis vertikal (||).

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

• Contoh salah

  Jika konsumen memiliki beberapa langganan ke topik, dan setiap langganan memiliki tag yang berbeda, konsumen hanya akan menerima pesan yang tag-nya cocok dengan tag yang ditentukan dalam langganan terbaru.

      // Dalam kode berikut, konsumen dapat menerima pesan dengan TagB dalam MQ_TOPIC tetapi tidak dapat menerima pesan dengan 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;
          }
      });                

Kode Contoh

• Mengirim pesan

  Konfigurasikan tag dan atribut kustom untuk pesan.

  Producer producer = ONSFactory.createProducer(properties);
  // Atur nilai Tag menjadi tagA.
  Message msg = new Message("topicA", "tagA", "Hello MQ".getBytes());
  // Atur nilai atribut kustom region menjadi hangzhou.
  msg.putUserProperties("region", "hangzhou");
  // Atur nilai atribut kustom price menjadi 50.
  msg.putUserProperties("price", "50");
  SendResult sendResult = producer.send(msg);

• Berlangganan pesan berdasarkan atribut kustom.

  Consumer consumer = ONSFactory.createConsumer(properties);
  // Berlangganan hanya pesan yang nilai atribut kustom region-nya adalah hangzhou. Jika Anda tidak mengonfigurasi atribut ini untuk pesan atau nilai atribut pesan bukan hangzhou, pesan tidak dikirimkan ke konsumen.
  consumer.subscribe("topicA", MessageSelector.bySql("region IS NOT NULL AND region = 'hangzhou'"), new MessageListener() {
      public Action consume(Message message, ConsumeContext context) {
          System.out.printf("Menerima Pesan Baru: %s %n", message);
          return Action.CommitMessage;
      }
  });

  Hasil yang diharapkan: Pesan yang dikirim dalam contoh memiliki atribut kustom region dan nilai atributnya adalah hangzhou. Pesan memenuhi kondisi filter dan dikirimkan ke konsumen.

• Berlangganan pesan berdasarkan tag dan atribut kustom.

  Consumer consumer = ONSFactory.createConsumer(properties);
  // Berlangganan hanya pesan yang memiliki tagA dan nilai atribut kustom price lebih besar dari 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("Menerima Pesan Baru: %s %n", message);
          return Action.CommitMessage;
      }
  });

  Hasil yang diharapkan: Pesan yang dikirim dalam contoh memiliki tagA dan atribut kustom price, dan nilai atribut kustom price lebih besar dari 30. Pesan memenuhi kondisi filter dan dikirimkan ke konsumen.

• Berlangganan pesan berdasarkan beberapa atribut kustom.

  Consumer consumer = ONSFactory.createConsumer(properties);
  // Berlangganan hanya pesan yang nilai atribut kustom region-nya adalah hangzhou dan nilai atribut kustom price kurang dari 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("Menerima Pesan Baru: %s %n", message);
          return Action.CommitMessage;
      }
  });

  Hasil yang diharapkan: Pesan tidak dikirimkan ke konsumen karena pesan tidak memenuhi kondisi filter. Konsumen berlangganan pesan yang nilai atribut kustom price-nya kurang dari 20, tetapi nilai atribut kustom price yang ditentukan untuk pesan dalam produsen adalah 50.

• Berlangganan semua pesan dalam topik.

  Consumer consumer = ONSFactory.createConsumer(properties);
  // Untuk berlangganan semua pesan dalam topik, atur nilai ekspresi SQL menjadi TRUE.
  consumer.subscribe("topicA", MessageSelector.bySql("TRUE"), new MessageListener() {
      public Action consume(Message message, ConsumeContext context) {
          System.out.printf("Menerima Pesan Baru: %s %n", message);
          return Action.CommitMessage;
      }
  });

  Hasil yang diharapkan: Semua pesan dalam topik dikirimkan ke konsumen.

• Contoh salah

  Sebelum produsen mengirimkan pesan, atribut kustom tidak dikonfigurasikan untuk pesan, dan tidak ada logika yang didefinisikan dalam ekspresi filter untuk memeriksa apakah atribut kustom ada. Atribut kustom langsung digunakan sebagai kondisi filter dalam ekspresi. Dalam hal ini, pesan tidak dikirimkan ke konsumen.

  Consumer consumer = ONSFactory.createConsumer(properties);
  // Atribut pesan produk tidak dikonfigurasikan selama pengiriman pesan. Penyaringan gagal dan pesan tidak dikirimkan ke konsumen.
  consumer.subscribe("topicA", MessageSelector.bySql("product = 'MQ'"), new MessageListener() {
      public Action consume(Message message, ConsumeContext context) {
          System.out.printf("Menerima Pesan Baru: %s %n", message);
          return Action.CommitMessage;
      }
  });