ActiveMQ（05）：JMS的API结构、开发步骤与Topic

一、JMS的API结构

二、一个JMS应用的基本步骤

1：创建一个JMS connection factory

2：通过connection factory来创建JMS connection

3：启动JMS connection

4：通过connection创建JMS session

5：创建JMS destination

6：创建JMS producer,或者创建JMS message，并设置destination

7：创建JMS consumer，或者是注册一个JMS message listener

8：发送或者接受JMS message(s)

9：关闭所有的JMS资源(connection, session, producer, consumer等)

三、topic

3.1 关于持久化和非持久化消息

持久化消息：

    这是ActiveMQ的默认传送模式，此模式保证这些消息只被传送一次和成功使用一次。对于这些消息，可靠性是优先考虑的因素。

  可靠性的另一个重要方面是确保持久性消息传送至目标后，消息服务在向消费者传送它们之前不会丢失这些消息。

   这意味着在持久性消息传送至目标时，消息服务将其放入持久性数据存储。如果消息服务由于某种原因导致失败，它可以恢复

  此消息并将此消息传送至相应的消费者。虽然这样增加了消息传送的开销，但却增加了可靠性。

非持久化消息：

    保证这些消息最多被传送一次。对于这些消息，可靠性并非主要的考虑因素。 此模式并不要求持久性的数据存储，也不保证

  消息服务由于某种原因导致失败后消息不会丢失。 有两种方法指定传送模式：

  1．使用setDeliveryMode 方法，这样所有的消息都采用此传送模式； 如：

    producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

  2．使用send 方法为每一条消息设置传送模式

3.2 非持久的Topic消息

3.2.1 消息的发送

基本跟前面发送队列信息是一样的，只是把创建Destination的地方，由创建队列替换成创建Topic，例如：

Destination destination = session.createTopic("my-topic");

3.2.1 消息的接收

1：必须要接收方在线，然后客户端再发送信息，接收方才能接收到消息

2：同样把创建Destination的地方，由创建队列替换成创建Topic，例如：

Destination destination = session.createTopic("my-topic");

3：由于不知道客户端发送多少信息，因此改成while循环的方式了，例如：

Message message = consumer.receive(); while(message!=null) {     TextMessage txtMsg = (TextMessage)message;     System.out.println("收到消 息：" + txtMsg.getText());     message = consumer.receive(1000L); }

3.3 持久的Topic消息

3.3.1 消息的发送

/**持久的*/ public void test2() throws Exception {     ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("liuy","123456","tcp://192.168.91.8:61616");     Connection connection = connectionFactory.createConnection();          Session session = connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);       Destination destination = session.createTopic("my-topic");          MessageProducer producer = session.createProducer(destination);     producer.setDeliveryMode(DeliveryMode.PERSISTENT); // 设置DeliveryMode.PERSISTENT模式     connection.start();     for (int i = 0; i < 3; i++) {         TextMessage message = session.createTextMessage("message22--" + i); producer.send(message);     }     session.commit();     session.close();     connection.close(); }

1：要用持久化订阅，发送消息者要用DeliveryMode.PERSISTENT模式发现，在连接之前设定

2：一定要设置完成后，再start这个connection

3.3.2 消息的接收

public void test2() throws Exception {     ConnectionFactory cf = new ActiveMQConnectionFactory("liuy","123456","tcp://192.168.91.8:61616");     Connection connection = cf.createConnection();     connection.setClientID("cc1");     final Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);     Topic topic = session.createTopic("my-topic");     TopicSubscriber ts = session.createDurableSubscriber(topic, "T1");     connection.start();     Message message = ts.receive();     while(message!=null) {         TextMessage txtMsg = (TextMessage)message; System.out.println("收到消息：" + txtMsg.getText()); message = ts.receive(1000L);     }     session.commit();     session.close();     connection.close(); }

1：需要在连接上设置消费者id，用来识别消费者

2：需要创建TopicSubscriber来订阅

3：要设置好了过后再start 这个 connection

4：一定要先运行一次，等于向消息服务中间件注册这个消费者，然后再运行客户端发送信息，这个时候，无论消费者是否在线，

  都会接收到，不在线的话，下次连接的时候，会把没有收过的消息都接收下来。