[译]JMS 发布-订阅消息模型

翻译 3 分钟阅读

在本文中，您将了解 JMS 发布-订阅 (publish-subscribe) 消息传递模型。正如您在JMS 简介文章中所读到的，在发布/订阅模型中，客户端通过称为主题的中介将消息发送给多个接收者。发送者通常称为发布者，接收者称为订阅者。

JMS 发布/订阅消息传递示例

下面是一个简单的代码示例，演示了发布/订阅消息模型的工作原理。我创建了 2 个主线程，publisher和subscriber1。将订阅者 1 克隆到subscriber2。所以基本上我有一个消息发布者和 2 个消息订阅者。链接至GitHub 。

package lab03.message.pubsub;


import labxx.common.settings.CommonSettings;
import javax.jms.*;


public class SimplePubSubExample {
  private static ConnectionFactory connectionFactory = null;
  private static Topic defaultTopic = null;
  static {
    connectionFactory = CommonSettings.getConnectionFactory();
    defaultTopic = CommonSettings.getDefautTopic();
  }


  public static void main(String[] args) {
    Thread publisher = new Thread(){
      @Override
      public void run(){
        try(JMSContext jmsContext = connectionFactory.createContext()) {
          Thread.sleep(1000);
          JMSProducer producer = jmsContext.createProducer();
          TextMessage message = jmsContext.createTextMessage("World needs to worry about the Climate changes");
          producer.send(defaultTopic, message);
        } catch (InterruptedException ex){
          ex.printStackTrace();
        }
      }
    };


    Thread subscriber1 = new Thread(){
      @Override
      public void run(){
        try(JMSContext jmsContext = connectionFactory.createContext()) {
          JMSConsumer consumer = jmsContext.createConsumer(defaultTopic);
          System.out.println("Message received: " + consumer.receive().getBody(String.class));
        } catch (JMSException e){
          e.printStackTrace();
        }
      }
    };


    Thread subscriber2 = new Thread(subscriber1);
    publisher.start();
    subscriber1.start();
    subscriber2.start();
  }
}
package labxx.common.settings;


import javax.jms.ConnectionFactory;
import javax.jms.Queue;
import javax.jms.Topic;
import javax.naming.InitialContext;
import javax.naming.NamingException;


public class CommonSettings {
  private static ConnectionFactory CONNECTION_FACTORY = null;
  private static Queue PTP_QUEUE = null;
  private static Topic PUB_SUB_TOPIC = null;
  private static Queue DEFAULT_REPLY_QUEUE = null;


  static {
    try {
      InitialContext initialContext = new InitialContext();
      CONNECTION_FACTORY = (ConnectionFactory) initialContext.lookup("jms/__defaultConnectionFactory");
      PTP_QUEUE = (Queue) initialContext.lookup("jms/PTPQueue");
      DEFAULT_REPLY_QUEUE = (Queue) initialContext.lookup("jms/ReplyQueue");
      PUB_SUB_TOPIC = (Topic) initialContext.lookup("jms/PubSubTopic");
    } catch (NamingException e) {
      e.printStackTrace();
    }
  }


  public static ConnectionFactory getConnectionFactory() {
    return CONNECTION_FACTORY;
  }
  public static Queue getDefaultQueue() {
    return PTP_QUEUE;
  }
  public static Queue getDefaultReplyQueue() {
    return DEFAULT_REPLY_QUEUE;
  }
  public static Topic getDefautTopic() {
    return PUB_SUB_TOPIC;
  }
}

输出

Message received: World needs to worry about the Climate changes
Message received: World needs to worry about the Climate changes

注意： 所有发布/订阅系统通常都存在一定延迟，您编写的代码应该将延迟考虑在内。

普通消息订阅者的问题

普通消息订阅者（如上例中使用的订阅者）的问题在于，它不耐用。也就是说，如果消费者因某种原因宕机（关闭），则在恢复后它将无法接收先前的消息Topic。

让我们看看下面的代码，发布者发送了 7 条消息，而消费者只收到一条消息。一旦消费者关闭并重新创建，它就不会从主题中获取消息。链接至**GitHub**。

package lab03.message.pubsub;


import labxx.common.settings.CommonSettings;
import javax.jms.*;


public class NormalConsumerProblem {
  private static ConnectionFactory connectionFactory = null;
  private static Topic defaultTopic = null;


  static {
    connectionFactory = CommonSettings.getConnectionFactory();
    defaultTopic = CommonSettings.getDefautTopic();
  }


  public static void main(String[] args) {
    Thread publisher = new Thread() {
      @Override
      public void run() {
        try (JMSContext jmsContext = connectionFactory.createContext()) {
          JMSProducer producer = jmsContext.createProducer();
          Thread.sleep(1000);
          for (int i = 1; i < 7; i++) {
            producer.send(defaultTopic, "Update " + i);
          }
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
    };


    //Normal Consumer
    Thread consumer = new Thread() {
      @Override
      public void run() {
        try (JMSContext jmsContext = connectionFactory.createContext()) {
          JMSConsumer consumer = jmsContext.createConsumer(defaultTopic);
          System.out.println(consumer.receive().getBody(String.class));
          Thread.sleep(2000);
          consumer.close();
          consumer = jmsContext.createConsumer(defaultTopic);
          for (int i = 1; i < 6; i++) {
            System.out.println(consumer.receive().getBody(String.class));
          }
        } catch (JMSException | InterruptedException e) {
          e.printStackTrace();
        }
      }
    };
    publisher.start();
    consumer.start();
  }
}

输出

Update 1

在实际场景中，您需要一种更可靠的方式来订阅主题。JMS 可以DurableConsumer解决这个问题。

JMS 发布-订阅中的 DurableConsumer 示例

下面的代码演示了 的用法DurableConsumer。与前面的示例不同，此处的消费者从 接收所有消息topic。链接至GitHub。

package lab03.message.pubsub;


import labxx.common.settings.CommonSettings;
import javax.jms.*;


public class DurableConsumerExample {
  private static ConnectionFactory connectionFactory = null;
  private static Topic defaultTopic = null;


  static {
    connectionFactory = CommonSettings.getConnectionFactory();
    defaultTopic = CommonSettings.getDefautTopic();
  }


  public static void main(String[] args) {
    Thread publisher = new Thread() {
      @Override
      public void run() {
        try (JMSContext jmsContext = connectionFactory.createContext()) {
          JMSProducer producer = jmsContext.createProducer();
          Thread.sleep(1000);
          for (int i = 1; i < 7; i++) {
            producer.send(defaultTopic, "Update " + i);
          }
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
    };


    //Durable Consumer
    Thread durableConsumer = new Thread() {
      @Override
      public void run() {
        try (JMSContext jmsContext = connectionFactory.createContext()) {
          jmsContext.setClientID("exampleApp");/important
          JMSConsumer consumer = jmsContext.createDurableConsumer(defaultTopic, "logConsumer");
          System.out.println(consumer.receive().getBody(String.class));
          Thread.sleep(2000);
          consumer.close();
          consumer = jmsContext.createDurableConsumer(defaultTopic, "logConsumer");
          for (int i = 1; i < 6; i++) {
            System.out.println(consumer.receive().getBody(String.class));
          }
        } catch (JMSException | InterruptedException e) {
          e.printStackTrace();
        }
      }
    };


    publisher.start();
    durableConsumer.start();
  }
}

输出

Update 1
Update 2
Update 3
Update 4
Update 5
Update 6

注意： 设置 ClientID 很重要，这通常是订阅客户端应用程序的名称。另外，请记住设置名称，durableConsumer如第 38 行和第 42 行所示。

SharedConsumer 进行负载平衡

在 JMS 2.0 中，多个订阅者可以监听一个主题，并且可以分配消息消费任务。有时，您可能需要对来自一个主题的多条消息进行负载平衡。这时您可以使用它来SharedConsumer分配消息消费负载。

下面的示例显示了创建sharedConsumer，请记住分配与第 37 行、第 38 行所示的相同的订阅者名称。代码链接至 GitHub。

package lab03.message.pubsub;


import labxx.common.settings.CommonSettings;
import javax.jms.*;


public class SharedConsumerExample {
  private static ConnectionFactory connectionFactory = null;
  private static Topic defaultTopic = null;


  static {
    connectionFactory = CommonSettings.getConnectionFactory();
    defaultTopic = CommonSettings.getDefautTopic();
  }


  public static void main(String[] args) {
    Thread publisher = new Thread() {
      @Override
      public void run() {
        try (JMSContext jmsContext = connectionFactory.createContext()) {
          JMSProducer producer = jmsContext.createProducer();
          Thread.sleep(1000);
          for (int i = 1; i < 7; i++) {
            producer.send(defaultTopic, "Update " + i);
          }
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
    };


    //Shared Consumer
    Thread sharedConsumer = new Thread() {
      @Override
      public void run() {
        try (JMSContext jmsContext = connectionFactory.createContext()) {
          JMSConsumer sharedConsumer1 = jmsContext.createSharedConsumer(defaultTopic, "sharedSubscriber");
          JMSConsumer sharedConsumer2 = jmsContext.createSharedConsumer(defaultTopic, "sharedSubscriber");
          for (int i = 0; i < 3; i++) {
            System.out.println("Shared Consumer1: " + sharedConsumer1.receive().getBody(String.class));
            System.out.println("Shared Consumer2: " + sharedConsumer2.receive().getBody(String.class));
          }
          Thread.sleep(3000);
          sharedConsumer1.close();
          sharedConsumer2.close();
        } catch (JMSException | InterruptedException e) {
          e.printStackTrace();
        }
      }
    };


    publisher.start();
    sharedConsumer.start();
  }
}

输出

Shared Consumer1: Update 1
Shared Consumer2: Update 2
Shared Consumer1: Update 3
Shared Consumer2: Update 4
Shared Consumer1: Update 5
Shared Consumer2: Update 6

此外，您还可以使用 ，SharedDurableConsumer它能为您提供DurableConsumer和 的综合能力SharedConsumer。

异步消息订阅

我在 JMS 点对点消息传递 文章中讨论了异步订阅。您可以使用相同的侦听器模式异步订阅主题。以下代码显示了异步消息订阅的一个简单示例。代码链接至GitHub。

package lab03.message.pubsub;


import labxx.common.settings.CommonSettings;
import javax.jms.*;


public class AsyncPubSubExample {
  private static ConnectionFactory connectionFactory = null;
  private static Topic defaultTopic = null;


  static {
    connectionFactory = CommonSettings.getConnectionFactory();
    defaultTopic = CommonSettings.getDefautTopic();
  }


  public static void main(String[] args) throws InterruptedException {
    try (JMSContext jmsContext = connectionFactory.createContext()) {
      JMSProducer producer = jmsContext.createProducer();
      JMSConsumer consumer = jmsContext.createConsumer(defaultTopic);
      consumer.setMessageListener(msg -> {
        try {
          System.out.println(msg.getBody(String.class));
        } catch (JMSException e) {
          e.printStackTrace();
        }
      });
      for (int i = 1; i < 7; i++) {
        producer.send(defaultTopic, "Message " + i);
      }
      Thread.sleep(1000);
      consumer.close();
    }
  }
}

输出

Message 1
Message 2
Message 3
Message 4
Message 5
Message 6

这些都是 JMS 发布-订阅消息模型的一部分，您已经清楚地了解了jms 发布/订阅中主题的使用。您了解了不同的消费者，例如 SharedConsumer、SharedDurableConsumer、DurableConsumer以及异步消息处理。请在下面的评论中分享您的想法。

原文链接：JMS pub-sub messaging model

订阅文章

订阅更新，不错过后续文章

直接通过 RSS 和 Telegram 订阅本站更新。

订阅 RSS关注 Telegram

分享文章

如果这篇有帮助，可以顺手转发

直接分享给同事、朋友，或者发到你的社交平台。

分享到 X 分享到 Telegram 邮件分享

相关文章

翻译

[译]JMS 2.0 中的新增功能（第一部分）- 易于使用

原文链接：https://www.oracle.com/technical-resources/articles/java/jms20.html ...

翻译

[译]JMS 2.0 中的新增功能（第二部分）- 新的消息传送功能

原文链接：https://www.oracle.com/technical-resources/articles/java/jms2messaging.html ...

翻译

[译]JMS 2.0 的十个好处意味着可以减少代码编写量

原文链接：Ten ways in which JMS 2.0 means writing less code ...

翻译

[译]JMS 2.0 中的新增功能

原文链接：What is new in JMS 2.0 ...

翻译

[译]完整的 JMS 2.0 教程

JMS 代表 Java 消息服务，它是 Java 中访问消息中间件服务器的标准。消息传递是在各种应用程序或服务之间交换业务数据的过程。这是一个完整的 JMS 2.0 教程， 其中包含大量代码示例。完整的代码示例可在 GitHub 上找到，请下载并运行示例。本教程使用 Glassfish 开源服务器作为 JMS 提供程序的示例。 ...

翻译

[译]使用 JMS 消息确认保证传送

JMS 通过 JMS 消息确认（确认模式）确保消息传递的可靠性。如果会话已进行事务处理，则该commit()方法会自动处理消息确认。否则，该方法将处理恢复rollback()。在本文中，我们将假设会话未进行事务处理。这是一个高级主题，请确保您对 JMS 有基本的了解。如果您是 JMS 新手，可以参考使用 JMS[发送和接收消息。](https://jstobigdata.com/jms/s ...

[译]JMS 消息优先级
[译]JMS 事务的实际应用