JMS 开发步骤、持久化 topic 消息与非持久化 topic 消息

女爷i 2024-04-19 06:50 84阅读 0赞

**目录**

JMS API 结构与开发步骤

非持久 Topic 消息

持久化 Topic 消息

--------------------

# JMS API 结构与开发步骤 #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdteDE5OTMzMjg_size_16_color_FFFFFF_t_70][]

JMS 开发的基本步骤如下：

> 1、创建一个 JMS Connection Factory
> 
> 2、通过 Connection Factory 创建 JMS Connection
> 
> 3、启动 JMS Connection
> 
> 4、通过 JMS Connection 创建 JMS Session
> 
> 5、创建 JMS Destination（目的地）
> 
> 6、创建 JMS Producer，或者创建 JMS Message，并设置 destination
> 
> 7、创建 JMS Consumer，或者注册一个 JMS message listener
> 
> 8、发送或者接收 JMS message
> 
> 9、关闭所有的 JMS 资源(connection、session、producer、consumer等)

下面通过具体的代码进行演示，分别介绍持久化与非持久化的 topic ，环境：Maven 3.6.1 + Java JDK 1.8 + ActiveMQ 5.15.9 + IDEA 2018 ，Mava 管理的 Java SE 应用，它们的 pom.xml 文件都是一样的，导入 activemq-all 依赖即可：

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0"
             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.wmx</groupId>
    <artifactId>activeMQ1</artifactId>
    <version>1.0-SNAPSHOT</version>
    <dependencies>
        <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-all</artifactId>
            <version>5.15.9</version>
        </dependency>
    </dependencies>
    </project>

《[ActiveMQ 简介 与 Maven 项目基本使用][ActiveMQ _ _ Maven]》中已经介绍过消息队列 queue，即使消费者没有在运行中，生产者发送的消息，都会在消费者下次启动时获取到，所以这里持久化与非持久化主要针对主题 topic。

# 非持久 Topic 消息 #

1、对于非持久化的 Topic 消息的发送：基本与发送队列一样，只需要把创建 Destination 的地方，由创建队列(queue)改成 Topic，例如：Destination destination = session.createTopic("topic-app");

2、对于非持久的 Topic 消息的接收：

> 1）必须要求接收方在线，然后客户端再发送消息，接收方才能接收到消息。
> 
> 2）同样把创建 Destination 的地方，由创建 queue 改为创建 topic。
> 
> 3）由于不知道客户端发送了多少 topic 消息，因此可以改为 while 循环的方式接收。
> 
> 4）同一个 topic ，某个消费者接收一次后，它不能再消费此消息，其它未消费的客户端可以继续接收此消息。

3、非持久化消息生产者将消息发送到中间件后，无论消息有没有被消费，默认情况下，再次启动中间件时，消息就没有了。

生产者代码如下：

import org.apache.activemq.ActiveMQConnectionFactory;
    import javax.jms.*;
    import java.util.UUID;
    /**
     * 非持久化 topic(主题) 消息生产者
     */
    @SuppressWarnings("all")
    public class NoPersistenceTopicSender {
        public static void main(String[] args) {
            Connection connection = null;
            Session session = null;
            MessageProducer messageProducer = null;
            try {
                String brokerURL = "tcp://127.0.0.1:61616";//ActiveMQ 中间件连接地址
                //创建 javax.jms.ConnectionFactory 连接工厂
                ConnectionFactory mqConnectionFactory = new ActiveMQConnectionFactory(brokerURL);
                //如果 ActiveMQ 连不上，则抛异常：java.net.ConnectException: Connection refused: connect
                connection = mqConnectionFactory.createConnection();//通过连接工厂获取连接 javax.jms.Connection
                connection.start();//启动连接，同理还有 stop、close
                //创建 session 会话，设置开启事务，消息确认模式为自动确认
                session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
                //创建消息主题(topic)，主题名称自己定义。Queue 、Topic 都是 Destination 的字接口
                Destination destination = session.createTopic("topic-app");
                messageProducer = session.createProducer(destination);//根据目的地创建消息生产者
                int massageTotal = 5;
                for (int i = 0; i < massageTotal; i++) {
                    TextMessage textMessage = session.createTextMessage("密文" + (i + 1) + ":" + UUID.randomUUID());
                    messageProducer.send(textMessage);//生产者发送消息
                }
                session.commit();//批量会话提交。此时消息会被正式发送到中间件
                System.out.println("消息发送完毕...");
            } catch (JMSException e) {
                e.printStackTrace();
            } finally {
                try {
                    if (session != null) {
                        session.close();//关闭会话
                    }
                    if (connection != null) {
                        connection.close();//关闭连接
                    }
                    if (messageProducer != null) {
                        messageProducer.close();//关闭生产者
                    }
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        }
    }

消费者代码如下：

import org.apache.activemq.ActiveMQConnectionFactory;
    import javax.jms.*;
    /**
     * 非持久化 topic(主题) 消息消费者
     */
    @SuppressWarnings("all")
    public class NoPersistenceTopicReceiver {
        public static void main(String[] args) {
            Connection connection = null;
            Session session = null;
            MessageConsumer messageConsumer = null;
            try {
                String brokerURL = "tcp://127.0.0.1:61616";//ActiveMQ 中间件连接地址
                ConnectionFactory mqConnectionFactory = new ActiveMQConnectionFactory(brokerURL);
                //如果 ActiveMQ 连不上，则抛异常：java.net.ConnectException: Connection refused: connect
                connection = mqConnectionFactory.createConnection();//通过连接工厂获取连接 javax.jms.Connection
                connection.start();//启动连接，同理还有 stop、close
                //创建会话 session。开启事务，消息确认模式为自动确认
                session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
                //创建消息主题(topic)，主题名称与生产者设置的保持一致。Queue 、Topic 都是 Destination 的字接口
                Destination destination = session.createTopic("topic-app");
                messageConsumer = session.createConsumer(destination);//根据目的地创建消息消费者
                Message message = messageConsumer.receive();//receive方法会导致当前线程阻塞，直到接收到消息
                while (message != null) {
                    if (message instanceof TextMessage) {
                        TextMessage textMessage = (TextMessage) message;
                        System.out.println("收到消息：" + textMessage.getText());
                        session.commit();//确认消息。
                        //接收消息时设置超时时间，单位为毫秒。如果为0，则等同于 receive()一致阻塞。
                        //如果超过超时时间，仍然未接收到消息，则返回 null。while 会推出
                        message = messageConsumer.receive(3000);
                    }
                }
            } catch (JMSException e) {
                e.printStackTrace();
            } finally {
                try {
                    if (session != null) {
                        session.close();//关闭会话
                    }
                    if (connection != null) {
                        connection.close();//关闭连接
                    }
                    if (messageConsumer != null) {
                        messageConsumer.close();//关闭消费者
                    }
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        }
    }

启动 activeMQ 中间件，测试如下：

> 非持久化 topic 消息时，如果先运行生产者发送消息，再运行消费者接收消息，那么消费者是接收不到之前发送的消息的，只能接收订阅主题时间点之后的消息。

![20190915093534829.gif][]

# 持久化 Topic 消息 #

1、对于持久化 topic 的发送：

>     1）在非持久的基础上设置传递模式（JMSDeliveryMode）(缺省时，默认是非持久)：
> 
>     producer.setDeliveryMode(DeliveryMode.PERSISTENT); //设置所有 message 采用同样的传递模式  
>     Message.setJMSDeliveryMode(DeliveryMode.PERSISTENT); //为每个 messge 分别设置传递模式
> 
>     2）设置完生产者的传递模式后，再启动连接：connection.start();
> 
>     3）对于发送的消息，activeMQ 默认情况下，不管消息有没有被投递/消费，activeMQ 重启后，消息就都没有了。

2、对于持久化 topic 的接收：

> 1）需要在连接上设置消费者 id，用于标识消费者：connection.setClientID("clientID\_100");
> 
> 2）创建 TopicSubscriber 来订阅，而不再是 MessageConsumer：session.createDurableSubscriber(destination,"ds\_1");
> 
> 3）设置好主题订阅后，再启动连接：connection.start();
> 
> 4）订阅者需要先运行一次，表示向消息中间件注册自己，然后运行客户端发送消息，此时无论订阅者是否在线，都能接收到，即使当时订阅者不在线，下次启动连接的时候，也会把没有收到的消息都接收下来。
> 
> 5）对于注册成功了的持久化订阅者，ActiveMQ 默认情况下，重启之后，仍然保存了注册了的订阅者，无需再次注册。

import org.apache.activemq.ActiveMQConnectionFactory;
    import javax.jms.*;
    import java.util.UUID;
    /**
     * 持久化 topic(主题) 消息生产者
     */
    @SuppressWarnings("all")
    public class PersistenceTopicSender {
        public static void main(String[] args) {
            Connection connection = null;
            Session session = null;
            MessageProducer messageProducer = null;
            try {
                String brokerURL = "tcp://127.0.0.1:61616";//ActiveMQ 中间件连接地址
                //创建 javax.jms.ConnectionFactory 连接工厂
                ConnectionFactory mqConnectionFactory = new ActiveMQConnectionFactory(brokerURL);
                //如果 ActiveMQ 连不上，则抛异常：java.net.ConnectException: Connection refused: connect
                connection = mqConnectionFactory.createConnection();//通过连接工厂获取连接 javax.jms.Connection
                //创建 session 会话，设置开启事务，消息确认模式为自动确认
                session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
                //创建消息主题(topic)，主题名称自己定义。Queue 、Topic 都是 Destination 的子接口
                Destination destination = session.createTopic("topic-app-2");
                messageProducer = session.createProducer(destination);//根据目的地创建消息生产者
                /**设置消息传递模式为持久化，不写时默认为非持久化。DeliveryMode.NON_PERSISTENT
                 * 设置完生产者的传递模式后，再启动连接：connection.start();
                 * */
                messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);
                connection.start();//启动连接，同理还有 stop、close
                int massageTotal = 5;
                for (int i = 0; i < massageTotal; i++) {
                    TextMessage textMessage = session.createTextMessage("密码" + (i + 1) + ":" + UUID.randomUUID());
                    messageProducer.send(textMessage);//生产者发送消息
                }
                session.commit();//批量会话提交。此时消息会被正式发送到中间件
                System.out.println("消息发送完毕...");
            } catch (JMSException e) {
                e.printStackTrace();
            } finally {
                try {
                    if (session != null) {
                        session.close();//关闭会话
                    }
                    if (connection != null) {
                        connection.close();//关闭连接
                    }
                    if (messageProducer != null) {
                        messageProducer.close();//关闭生产者
                    }
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        }
    }

import org.apache.activemq.ActiveMQConnectionFactory;
    import javax.jms.*;
    /**
     * 持久化 topic(主题) 消息消费者
     */
    @SuppressWarnings("all")
    public class PersistenceTopicReceiver {
        public static void main(String[] args) {
            Connection connection = null;
            Session session = null;
            TopicSubscriber topicSubscriber = null;
            try {
                String brokerURL = "tcp://127.0.0.1:61616";//ActiveMQ 中间件连接地址
                ConnectionFactory mqConnectionFactory = new ActiveMQConnectionFactory(brokerURL);
                //如果 ActiveMQ 连不上，则抛异常：java.net.ConnectException: Connection refused: connect
                connection = mqConnectionFactory.createConnection();//通过连接工厂获取连接 javax.jms.Connection
                /**为持久订阅设置客户端id，这样即使订阅者不在线，消息中心也能在它下次上线时将消息投递给它*/
                connection.setClientID("clientID_100");
                //创建会话 session。开启事务，消息确认模式为自动确认
                session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
                //创建消息主题(topic)，主题名称与生产者设置的保持一致。Queue 、Topic 都是 Destination 的子接口
                Topic topic = session.createTopic("topic-app-2");
                /**createDurableSubscriber(Topic var1, String var2)：创建持久订阅，var1 是订阅对象，var2 是持久订阅名称，自定义即可
                 * 主题订阅接口 TopicSubscriber 继承了 MessageConsumer 消息消费者接口*/
                topicSubscriber = session.createDurableSubscriber(topic,"ds-1");//根据目的地创建消息消费者
                connection.start();/**设置了主题订阅后，再启动连接*/
                System.out.println("订阅者启动成功...");
                Message message = topicSubscriber.receive();//receive方法会导致当前线程阻塞，直到接收到消息
                while (message != null) {
                    if (message instanceof TextMessage) {
                        TextMessage textMessage = (TextMessage) message;
                        System.out.println("收到消息：" + textMessage.getText());
                        session.commit();//确认消息。
                        //接收消息时设置超时时间，单位为毫秒。如果为0，则等同于 receive()一致阻塞。
                        //如果超过超时时间，仍然未接收到消息，则返回 null。while 会推出
                        message = topicSubscriber.receive(3000);
                    }
                }
            } catch (JMSException e) {
                e.printStackTrace();
            } finally {
                try {
                    if (session != null) {
                        session.close();//关闭会话
                    }
                    if (connection != null) {
                        connection.close();//关闭连接
                    }
                    if (topicSubscriber != null) {
                        topicSubscriber.close();//关闭消费者
                    }
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        }
    }

![2019091512094845.gif][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdteDE5OTMzMjg_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/493060db0fbc4f29b4025e1d1e4168c9.png
[ActiveMQ _ _ Maven]: https://mp.csdn.net/postedit/100623202
[20190915093534829.gif]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/68290c95a5864f2ead3409dbf59817ac.gif
[2019091512094845.gif]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/b2d04c5472ca40ff96fc9e51a746dcf2.gif