activemq message persistence

queue消息持久化

 queue默认是持久化的;When the producer,生产1000条消息;此时关闭mq服务,再开启mq服务,The background or display at this time1000Message to be processed;

queue设置为非持久化;代码如下;When the producer,生产1条消息;此时关闭mq服务,再开启mq服务,Background shows no news to be processed at this time;如下图所示：

package com.mock.utils;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Queue;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;

public class TestActiveMqProducerCanDelete {
    private static final String ACTIVEMQ_URL = "tcp://192.168.2.189:61616";
    private static final String QUEUE_NAME = "queue_01";
    private static final String QUEUE_NAME_2 = "queue_02";

    public static void main(String[] args) throws JMSException {
        // 创建连接工厂,按照给定的URL,The default user name password
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
        // 通过连接工厂 获取connection 并启动访问
        Connection conn = activeMQConnectionFactory.createConnection();
        conn.start();
        // 创建session会话
        Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 创建目的地 （具体是队列还是主题topic）
        Queue queue = session.createQueue(QUEUE_NAME);
        // 创建消息的生产者
        MessageProducer messageProducer = session.createProducer(queue);
        messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
        // Byte类型的数据
        ObjectMessage message = session.createObjectMessage();
        User user = new User();
        user.setAddress("嘉兴");
        user.setName("Joy");
        message.setObject(user);
        message.setStringProperty("StringProperty", "我是 属性xxxxxxx");
        messageProducer.send(message);

        messageProducer.close();
        session.close();
        conn.close();
        System.out.println("发送消息成功");
    }

}

 

 

 

 

topic消息持久化:

topic 默认是不持久化的.或者说topicThe persistence of need to modify the code; 在hello world中的实现方式,The message will not be persisted;

• topic消息持久化,Like attention WeChat public number,If the exit WeChat after,仍然可以接收To the public on the client side, quit after some push;
• topic消息非持久化：Like attention WeChat public number,If the exit WeChat after,无法接收Went to the public in the exit this time push;If the device is online again,The offline this time don't push so;At this time if online again,This time can only receive online push;

消费者代码：Create a persistent subscribers,clientId为 xx;监听消息;

package com.mock.utils;

import java.io.IOException;

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.TopicSubscriber;

import org.apache.activemq.ActiveMQConnectionFactory;

public class TestActiveMqTopicConsumer {
    private static final String ACTIVEMQ_URL = "tcp://192.168.2.189:61616";
    private static final String TOPIC_NAME = "TOPIC_NAME_1";

    public static void main(String[] args) throws JMSException, IOException {
        System.out.println("****我是1号消费者******");
        // 创建连接工厂,按照给定的URL,The default user name password
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
        // 通过连接工厂 获取connection 并启动访问
        Connection conn = activeMQConnectionFactory.createConnection();
        conn.setClientID("xx");
        // 创建session会话
        Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 创建目的地 （具体是队列还是主题topic）
        Topic topic = session.createTopic(TOPIC_NAME);
        TopicSubscriber topicSubscriber = session.createDurableSubscriber(topic, "xx_xx");
        conn.start();
        topicSubscriber.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                if (message != null && message instanceof TextMessage) {
                    TextMessage textMessage = (TextMessage) message;
                    try {
                        System.out.println("收到消息：" + textMessage.getText());
                    } catch (JMSException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
            }

        });
        System.in.read();

        topicSubscriber.close();
        session.close();
        conn.close();

    }

}

生产者代码：

package com.mock.utils;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;

import org.apache.activemq.ActiveMQConnectionFactory;

public class TestActiveMqTopicProducer {
    private static final String ACTIVEMQ_URL = "tcp://192.168.2.189:61616";
    private static final String TOPIC_NAME = "TOPIC_NAME_1";

    public static void main(String[] args) throws JMSException {
        // 创建连接工厂,按照给定的URL,The default user name password
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
        // 通过连接工厂 获取connection 并启动访问
        Connection conn = activeMQConnectionFactory.createConnection();

        // 创建session会话
        Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 创建目的地 （具体是队列还是主题topic）
        Topic topic = session.createTopic(TOPIC_NAME);

        // 创建消息的生产者
        MessageProducer messageProducer = session.createProducer(topic);
        messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);
        conn.start();
        for (int i = 0; i < 3; i++) {
            // 创建消息;Can understand in accordance with the requirements for the students to write a good question
            TextMessage textMessage = session.createTextMessage("mession-------" + i);
            // 通过messageProducer 发送给mq
            messageProducer.send(textMessage);
        }
        messageProducer.close();
        session.close();
        conn.close();
        System.out.println("发送消息成功");
    }

}

由于JMSDeliveryMode默认是持久化的,所以,Producers code can also be so write：

public static void main(String[] args) throws JMSException {
        // 创建连接工厂,按照给定的URL,The default user name password
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
        // 通过连接工厂 获取connection 并启动访问
        Connection conn = activeMQConnectionFactory.createConnection();
        conn.start();
        // 创建session会话
        Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 创建目的地 （具体是队列还是主题topic）
        Topic topic = session.createTopic(TOPIC_NAME);

        // 创建消息的生产者
        MessageProducer messageProducer = session.createProducer(topic);

        for (int i = 0; i < 3; i++) {
            // 创建消息;Can understand in accordance with the requirements for the students to write a good question
            TextMessage textMessage = session.createTextMessage("mession-------" + i);
            // 通过messageProducer 发送给mq
            messageProducer.send(textMessage);
        }
        messageProducer.close();
        session.close();
        conn.close();
        System.out.println("发送消息成功");
    }

 

activemq Management background according to the diagram below,Start the consumer code,显示如下图;

启动消费者,显示如下图,An onlineTOPIC 订阅者;

If consumers closed this time,So the background shows the same foreverTopicThe subscriber offline;

In the permanentTopicThe subscriber offline cases,生产者生产了3条消息,Then the background management according 永久TopicThe subscriber has3条待处理的消息;