ActiveMQ中JMS的可靠性机制

　　全文用到的生产者代码:

package cn.qlq.activemq;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * 生产消息
 * 
 * @author QiaoLiQiang
 * @time 2018年9月18日下午11:04:41
 */
public class MsgProducer {

    // 默认端口61616
    private static final String url = "tcp://localhost:61616/";
    private static final String queueName = "myQueue";
    private static Session session = null;

    public static void main(String[] args) throws JMSException {
        // 1创建ConnectionFactory
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
        // 2.由connectionFactory创建connection
        Connection connection = connectionFactory.createConnection();
        // 3.启动connection
        connection.start();
        // 4.创建Session===第一个参数是是否事务管理，第二个参数是应答模式
        session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
        // 5.创建Destination(Queue继承Queue)
        Queue destination = session.createQueue(queueName);
        // 6.创建生产者producer
        MessageProducer producer = session.createProducer(destination);
        for (int i = 0; i < 10; i++) {
            // 7.创建Message，有好多类型，这里用最简单的TextMessage
            TextMessage tms = session.createTextMessage("textMessage:" + i);

            // 设置附加属性
            tms.setStringProperty("str", "stringProperties" + i);

            // 8.生产者发送消息
            producer.send(tms);
        }

        // 9.提交事务
        session.commit();

        // 10.关闭connection
        session.close();
        connection.close();
    }

}

 1.消息接收确认

　　JMS 消息只有在被确认之后，才认为已经被成功地消费了。消息的成功消费通常包含三个阶段：客户接收消息、客户处理消息和消息被确认。

(1)　在事务性会话中，事务被提交的时候，确认自动发生。也就是事务性回话需要进行session.commit()。如下消费者代码:

package cn.qlq.activemq;

import java.util.Enumeration;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * 消费消息
 * 
 * @author QiaoLiQiang
 * @time 2018年9月18日下午11:26:41
 */
public class MsgConsumer {

    // 默认端口61616
    private static final String url = "tcp://localhost:61616/";
    private static final String queueName = "myQueue";

    public static void main(String[] args) throws JMSException {
        // 1创建ConnectionFactory
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
        // 2.由connectionFactory创建connection
        Connection connection = connectionFactory.createConnection();
        Enumeration jmsxPropertyNames = connection.getMetaData().getJMSXPropertyNames();
        while (jmsxPropertyNames.hasMoreElements()) {
            String nextElement = (String) jmsxPropertyNames.nextElement();
            System.out.println("JMSX name ===" + nextElement);
        }
        // 3.启动connection
        connection.start();
        // 4.创建Session===第一个参数是是否事务管理，第二个参数是应答模式
        Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
        // 5.创建Destination(Queue继承Queue)
        Queue destination = session.createQueue(queueName);
        // 6.创建消费者consumer
        MessageConsumer consumer = session.createConsumer(destination);

        int i = 0;
        while (i < 10) {
            TextMessage textMessage = (TextMessage) consumer.receive();
            System.out.println("接收消息:" + textMessage.getText() + ";属性" + textMessage.getStringProperty("str"));
            i++;

        }

        // 提交事务，进行确认收到消息
        session.commit();

        session.close();
        connection.close();
    }
}

补充:如果在事务性会话中应答模式是Session.SESSION_TRANSACTED，也需要手动提交事务:

        // 4.创建Session===第一个参数是是否事务管理，第二个参数是应答模式
        Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
　　　　　　...
        // 提交事务，进行确认收到消息
        session.commit();

(2)在非事务性会话中，消息何时被确认取决于创建会话的应答模式(acknowledgement mode)。该参数有以下三个值可选:

• Session.AUTO_ACKNOWLEDGE

　　当客户成功的从receive方法返回的时候，或者从 MessageListener.onMessage方法成功返回的时候，会话自动确认客户收到的消息。(接收端线程不会结束，原因是listener新起了一个线程一直监听，如果再次运行接收端新开一个consumer是接收不到消息的，消息已经被第一个消费了)

package cn.qlq.activemq;

import java.util.Enumeration;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * 消费消息
 * 
 * @author QiaoLiQiang
 * @time 2018年9月18日下午11:26:41
 */
public class MsgConsumer {

    // 默认端口61616
    private static final String url = "tcp://localhost:61616/";
    private static final String queueName = "myQueue";

    public static void main(String[] args) throws JMSException {
        // 1创建ConnectionFactory
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
        // 2.由connectionFactory创建connection
        Connection connection = connectionFactory.createConnection();
        Enumeration jmsxPropertyNames = connection.getMetaData().getJMSXPropertyNames();
        while (jmsxPropertyNames.hasMoreElements()) {
            String nextElement = (String) jmsxPropertyNames.nextElement();
            System.out.println("JMSX name ===" + nextElement);
        }
        // 3.启动connection
        connection.start();
        // 4.创建Session===第一个参数是是否事务管理，第二个参数是应答模式
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 5.创建Destination(Queue继承Queue)
        Queue destination = session.createQueue(queueName);
        // 6.创建消费者consumer
        MessageConsumer consumer = session.createConsumer(destination);

        int i = 0;
        while (i < 10) {
            TextMessage textMessage = (TextMessage) consumer.receive();
            System.out.println("接收消息:" + textMessage.getText() + ";属性" + textMessage.getStringProperty("str"));
            i++;

        }

        session.close();
        connection.close();
    }
}

• Session.CLIENT_ACKNOWLEDGE  客户端应答

　　客户通过调用消息的acknowledge方法确认消息。需要注意的是，在这种模式中，确认是在会话层上进行，确认一个被消费的消息将自动确认所有已被会话消费的消息。例如，如果一个消息消费者消费了10 个消息，然后确认第5 个消息，那么所有10 个消息都被确认。如果接受了五个消息然后进行了确认，又接受了五个消息，只会对前面五个进行确认，后面接受的消息不会影响。

package cn.qlq.activemq;

import java.util.Enumeration;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * 消费消息
 * 
 * @author QiaoLiQiang
 * @time 2018年9月18日下午11:26:41
 */
public class MsgConsumer {

    // 默认端口61616
    private static final String url = "tcp://localhost:61616/";
    private static final String queueName = "myQueue";

    public static void main(String[] args) throws JMSException {
        // 1创建ConnectionFactory
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
        // 2.由connectionFactory创建connection
        Connection connection = connectionFactory.createConnection();
        Enumeration jmsxPropertyNames = connection.getMetaData().getJMSXPropertyNames();
        while (jmsxPropertyNames.hasMoreElements()) {
            String nextElement = (String) jmsxPropertyNames.nextElement();
            System.out.println("JMSX name ===" + nextElement);
        }
        // 3.启动connection
        connection.start();
        // 4.创建Session===第一个参数是是否事务管理，第二个参数是应答模式
        Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
        // 5.创建Destination(Queue继承Queue)
        Queue destination = session.createQueue(queueName);
        // 6.创建消费者consumer
        MessageConsumer consumer = session.createConsumer(destination);

        int i = 0;
        while (i < 10) {
            TextMessage textMessage = (TextMessage) consumer.receive();
            System.out.println("接收消息:" + textMessage.getText() + ";属性" + textMessage.getStringProperty("str"));
            i++;

            if (i == 10) {// 确保消费完所有的消息再进行确认
                textMessage.acknowledge();
            }
        }

        session.close();
        connection.close();
    }
}

• Session.DUPS_OK_ACKNOWLEDGE (了解就好)

该选择只是会话迟钝的确认消息的提交。如果JMS provider失败，那么可能会导致一些重复的消息。如果是重复的消息，那么JMS provider 必须把消息头的JMSRedelivered字段设置为true

2.第二种机制

1.消息持久性

JMS 支持以下两种消息提交模式：
　　PERSISTENT：只是JMS provider持久保存消息，以保证消息不会因为JMS provider的失败而丢失
　　NON_PERSISTENT：不要求JMS provider持久保存消息

// producer.setDeliveryMode(DeliveryMode.PERSISTENT);将消息传递特性置为持久化
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);// 非持久化

2.消息优先级　　

　　可以使用消息优先级来指示JMS provider首先提交紧急的消息。优先级分10个级别，从0（最低）到9（最高）。如果不指定优先级，默认级别是4。需要注意的是，JMS provider并不一定保证按照优先级的顺序提交消息

3.消息过期

可以设置消息在一定时间后过期，默认是永不过期

producer.setTimeToLive(Long aliveTime);

　　注意timeToLive属性只会在DisableMessageTimestamp=false(禁用消息时间戳)的情况下才有意义

4.消息的临时目的地

　　可以通过会话上的createTemporaryQueue 方法和createTemporaryTopic 方法来创建临时目的地。它们的存在时间只限于创建它们的连接所保持的时间。只有创建该临时目的地的连接上的消息消费者才能够从临时目的地中提取消息。

生产者:

package cn.qlq.activemq;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TemporaryQueue;
import javax.jms.TemporaryTopic;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * 生产消息
 * 
 * @author QiaoLiQiang
 * @time 2018年9月18日下午11:04:41
 */
public class MsgProducer {

    // 默认端口61616
    private static final String url = "tcp://localhost:61616/";
    private static final String queueName = "myQueue";
    private static Session session = null;

    public static void main(String[] args) throws JMSException {
        // 1创建ConnectionFactory
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
        // 2.由connectionFactory创建connection
        Connection connection = connectionFactory.createConnection();
        // 3.启动connection
        connection.start();
        // 4.创建Session===第一个参数是是否事务管理，第二个参数是应答模式
        session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
        // 5.创建Destination(Queue继承Queue)
        Queue destination = session.createQueue(queueName);

        TemporaryQueue temporaryQueue = session.createTemporaryQueue();

        // 6.创建生产者producer
        MessageProducer producer = session.createProducer(destination);
        for (int i = 0; i < 1; i++) {
            // 7.创建Message，有好多类型，这里用最简单的TextMessage
            TextMessage tms = session.createTextMessage("textMessage:" + i);

            // 设置附加属性
            tms.setStringProperty("str", "stringProperties" + i);

            tms.setJMSReplyTo(temporaryQueue);

            // 8.生产者发送消息
            producer.send(tms);
        }

        // 9.提交事务
        session.commit();

        // 10.关闭connection
        session.close();
        connection.close();
    }

}

 执行一次查看消息后台:

 消费者:

package cn.qlq.activemq;

import java.util.Enumeration;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * 消费消息
 * 
 * @author QiaoLiQiang
 * @time 2018年9月18日下午11:26:41
 */
public class MsgConsumer {

    // 默认端口61616
    private static final String url = "tcp://localhost:61616/";
    private static final String queueName = "myQueue";

    public static void main(String[] args) throws JMSException {
        // 1创建ConnectionFactory
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
        // 2.由connectionFactory创建connection
        Connection connection = connectionFactory.createConnection();
        Enumeration jmsxPropertyNames = connection.getMetaData().getJMSXPropertyNames();
        while (jmsxPropertyNames.hasMoreElements()) {
            String nextElement = (String) jmsxPropertyNames.nextElement();
            System.out.println("JMSX name ===" + nextElement);
        }
        // 3.启动connection
        connection.start();
        // 4.创建Session===第一个参数是是否事务管理，第二个参数是应答模式
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 5.创建Destination(Queue继承Queue)
        Queue destination = session.createQueue(queueName);
        // 6.创建消费者consumer
        MessageConsumer consumer = session.createConsumer(destination);

        int i = 0;
        while (i < 1) {
            TextMessage textMessage = (TextMessage) consumer.receive();
            System.out.println("接收消息:" + textMessage.getText() + ";属性" + textMessage.getStringProperty("str"));
            i++;

            Destination jmsReplyTo = textMessage.getJMSReplyTo();
            System.out.println(jmsReplyTo);
        }

        session.close();
        connection.close();
    }
}

结果:

JMSX name ===JMSXUserID
JMSX name ===JMSXGroupID
JMSX name ===JMSXGroupSeq
JMSX name ===JMSXDeliveryCount
JMSX name ===JMSXProducerTXID
接收消息:textMessage:0;属性stringProperties0
temp-queue://ID:MicroWin10-1535-49479-1551765172839-1:1:1

3.机制3  ====持久订阅

 　　首先消息生产者必须使用PERSISTENT提交消息。客户可以通过会话上的createDurableSubscriber方法来创建一个持久订阅，该方法的第一个参数必须是一个topic。第二个参数是订阅的名称。JMS provider会存储发布到持久订阅对应的topic上的消息。如果最初创建持久订阅的客户或者任何其它客户，使用相同的连接工厂和连接的客户ID，相同的主题和相同的订阅名，再次调用会话上的createDurableSubscriber方法，那么该持久订阅就会被激活。JMS provider会向客户发送客户处于非激活状态时所发布的消息。持久订阅在某个时刻只能有一个激活的订阅者。持久订阅在创建之后会一直保留，直到应用程序调用会话上的unsubscribe方法。

4.  机制4

1.本地事务

　　在一个JMS客户端，可以使用本地事务来组合消息的发送和接收。

　　JMS Session接口提供了commit和rollback方法。事务提交意味着生产的所有消息被发送，消费的所有消息被确认；事务回滚意味着生产的所有消息被销毁，消费的所有消息被恢复并重新提交，除非它们已经过期。

　　事务性的会话总是牵涉到事务处理中，commit或rollback方法一旦被调用，一个事务就结束了，而另一个事务被开始。关闭事务性会话将回滚其中的事务。

　　需要注意的是，如果使用请求/回复机制，即发送一个消息，同时希望在同一个事务中等待接收该消息的回复，那么程序将被挂起，因为直到事务提交，发送操作才会真正执行。需要注意的还有一个，消息的生产和消费不能包含在同一个事务中。

2.JMS的PTP模型(队列模式)

　　JMS PTP(Point-to-Point)模型定义了客户端如何向队列发送消息，从队列接收消息，以及浏览队列中的消息。

　　PTP模型是基于队列的，生产者发消息到队列，消费者从队列接收消息，队列的存在使得消息的异步传输成为可能。和邮件系统中的邮箱一样，队列可以包含各种消息，JMS Provider 提供工具管理队列的创建、删除。

　　PTP的一些特点：

1：如果在Session 关闭时，有一些消息已经被收到，但还没有被签收(acknowledged)，那么，当消费者下次连接到相同的队列时，这些消息还会被再次接收
2：如果用户在receive 方法中设定了消息选择条件，那么不符合条件的消息会留在队列中，不会被接收到
3：队列可以长久地保存消息直到消费者收到消息。消费者不需要因为担心消息会丢失而时刻和队列保持激活的连接状态，充分体现了异步传输模式的优势

3.JMS的Pub/Sub模型(重要)

 　　JMS Pub/Sub 模型定义了如何向一个内容节点发布和订阅消息，这些节点被称作topic

　　主题可以被认为是消息的传输中介，发布者(publisher)发布消息到主题，订阅者(subscribe) 从主题订阅消息。主题使得消息订阅者和消息发布者保持互相独立，不需要接触即可保证消息的传送。

    Pub/Sub的一些特点：

1：消息订阅分为非持久订阅和持久订阅
　　　非持久订阅只有当客户端处于激活状态，也就是和JMS Provider保持连接状态才能收到发送到某个主题的消息，而当客户端处于离线状态，这个时间段发到主题的消息将会丢失，永远不会收到。
　　　持久订阅时，客户端向JMS 注册一个识别自己身份的ID，当这个客户端处于离线时，JMS Provider会为这个ID 保存所有发送到主题的消息，当客户再次连接到JMS Provider时，会根据自己的ID 得到所有当自己处于离线时发送到主题的消息。(必须先启动消费者注册一下ID)
2：如果用户在receive 方法中设定了消息选择条件，那么不符合条件的消息不会被接收
3：非持久订阅状态下，不能恢复或重新派送一个未签收的消息。只有持久订阅才能恢复或重新派送一个未签收的消息。
4：当所有的消息必须被接收，则用持久订阅。当丢失消息能够被容忍，则用非持久订阅

• 非持久订阅的生产者和消费者

public class MsgProducer {

    // 默认端口61616
    private static final String url = "tcp://localhost:61616/";
    private static final String topicName = "myTopic";

    public static void main(String[] args) throws JMSException {
        // 1创建ConnectionFactory
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
        // 2.由connectionFactory创建connection
        Connection connection = connectionFactory.createConnection();
        // 3.启动connection
        connection.start();
        // 4.创建Session===第一个参数是是否事务管理，第二个参数是应答模式
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 5.创建Destination(Queue继承Queue,Topic也继承Destination==这三个都是接口)
        Destination destination = session.createTopic(topicName);
        // 6.创建生产者producer
        MessageProducer producer = session.createProducer(destination);
        for (int i = 0; i < 100; i++) {
            // 7.创建Message，有好多类型，这里用最简单的TextMessage
            TextMessage tms = session.createTextMessage("textMessage:" + i);
            // 8.生产者发送消息
            producer.send(tms);

            System.out.println("send:" + tms.getText());
        }
        // 9.关闭connection
        connection.close();
    }

}

public class MsgConsumer {

    // 默认端口61616
    private static final String url = "tcp://localhost:61616/";
    private static final String topicName = "myTopic";

    public static void main(String[] args) throws JMSException {
        // 1创建ConnectionFactory
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
        // 2.由connectionFactory创建connection
        Connection connection = connectionFactory.createConnection();
        // 3.启动connection
        connection.start();
        // 4.创建Session===第一个参数是是否事务管理，第二个参数是应答模式
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 5.创建Destination(Queue继承Queue,Topic也继承Destination==这三个都是接口)
        Destination destination = session.createTopic(topicName);
        // 6.创建消费者consumer
        MessageConsumer consumer = session.createConsumer(destination);
        // 7.给消费者绑定监听器(消息的监听是一个异步的过程，不可以关闭连接，绑定监听器线程是一直开启的，处于阻塞状态，所以可以在程序退出关闭)
        consumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                // 7.1由于消费者接受的是TextMessage，所以强转一下
                TextMessage tms = (TextMessage) message;
                try {
                    System.out.println("接收消息:" + tms.getText());
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        });
    }

}

• 持久订阅的生产者和消费者

　　　　生产者将发布模式改为持久模式，默认是持久模式。必须在开启connection之前设置。

public class MsgProducer {

    // 默认端口61616
    private static final String url = "tcp://localhost:61616/";
    private static final String topicName = "myTopic";

    public static void main(String[] args) throws JMSException {
        // 1创建ConnectionFactory
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
        // 2.由connectionFactory创建connection
        Connection connection = connectionFactory.createConnection();

        // 3.创建Session===第一个参数是是否事务管理，第二个参数是应答模式
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 4.创建Destination(Queue继承Queue,Topic也继承Destination==这三个都是接口)
        Destination destination = session.createTopic(topicName);
        // 5.创建生产者producer
        MessageProducer producer = session.createProducer(destination);
        // 6设置为持久模式(这个必须在下面开启connection之前)
        producer.setDeliveryMode(DeliveryMode.PERSISTENT);
        // 7.启动connection
        connection.start();

        for (int i = 0; i < 10; i++) {
            // 8.创建Message，有好多类型，这里用最简单的TextMessage
            TextMessage tms = session.createTextMessage("textMessage:" + i);
            // 9.生产者发送消息
            producer.send(tms);

            System.out.println("send:" + tms.getText());
        }
        // 9.关闭connection
        connection.close();
    }
}

需要注意的是:　

　　1：要用持久化订阅，发送消息者要用 DeliveryMode.PERSISTENT 模式，在连接之前设定
　　2：一定要设置完成后，再start 这个 connection

public class MsgConsumer {

    // 默认端口61616
    private static final String url = "tcp://localhost:61616/";
    private static final String topicName = "myTopic";

    public static void main(String[] args) throws JMSException {
        // 1创建ConnectionFactory
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
        // 2.由connectionFactory创建connection
        Connection connection = connectionFactory.createConnection();
        // 设置链接的ID
        connection.setClientID("cc1");
        // 3.创建Session===第一个参数是是否事务管理，第二个参数是应答模式
        Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
        // 4.创建Destination(Queue继承Queue,Topic也继承Destination==这三个都是接口)
        Topic destination = session.createTopic(topicName);
        // 创建TopicSubscriber来订阅;需要在连接上设置消费者id，用来识别消费者;设置好了过后再start 这个 connection
        TopicSubscriber ts = session.createDurableSubscriber(destination, "C11");
        // 5.启动connection
        connection.start();
        // 6.接受消息(用上面的TopicSubscriber接收消息)
        Message message = ts.receive();
        while (message != null) {
            TextMessage txtMsg = (TextMessage) message;
            session.commit();
            System.out.println("收到消 息：" + txtMsg.getText());
            message = ts.receive(1000L);
        }
        session.close();
        connection.close();
    }
}

需要注意的是:

1：需要在连接上设置消费者id，用来识别消费者。也就是 connection.setClientID("cc1");
2：需要创建TopicSubscriber来订阅，也就是 session.createDurableSubscriber(destination, "C11");
3：要设置好了过后再start 这个 connection
4：一定要先运行一次，等于向消息服务中间件注册这个消费者，然后再运行客户端发送信息，这个时候，无论消费者是否在线，都会接收到，不在线的话，下次连接的时候，会把没有收过的消息都接收下来。
5: 接受消息是使用TopicSubscriber来接收消息

对于持久订阅我们查看activemq订阅的后台:

补充:关于持久化消息和费持久化消息

有两种方式指定传送模式:

　　1.使用setDeliveryMode方法，这样所有的消息都采用此传送模式;如producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT)

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

持久化消息

　　持久化消息是 ActiveMQ 的默认传送模式，此模式保证这些消息只被传送一次和成功使用一次。对于这些消息，可靠性是优先考虑的因素。可靠性的另一个重要方面是确保持久性消息传送至目标后，消息服务在向消费者传送它们之前不会丢失这些消息。

　　这意味着在持久性消息传送至目标时，消息服务将其放入持久性数据存储。如果消息服务由于某种原因导致失败，它可以恢复此消息并将此消息传送至相应的消费者。虽然这样增加了消息传送的开销，但却增加了可靠性。

非持久化消息

　　保证这些消息最多被传送一次。对于这些消息，可靠性并非主要的考虑因素。此模式并不要求持久性的数据存储，也不保证消息服务由于某种原因导致失败后消息不会丢失。