ActiveMQ 4.JMS规范与落地产品

叁歲伎倆 2023-05-30 03:38 35阅读 0赞

### 文章目录 ###

*  ActiveMQ
 *   *  四 JMS规范与落地产品
     *   *  1 是什么
         *   *  1.1 JavaEE
             *  1.2 Java Message Service(Java消息服务是JavaEE中的一个技术
         *  2 MQ中间件的其他落地产品
         *  3 JMS的组成结构和特点
         *   *  3.1 JMS Provider
             *  3.2 JMS producer
             *  3.3 JMS consumer
             *  3.4 JMS message
             *   *  3.4.1 消息头
                 *  3.4.2 消息体
                 *  3.4.3 消息属性
         *  4 JMS的可靠性
         *   *  4.1 持久化PERSISTENT
             *   *  4.1.1 参数设置说明
                 *  4.1.2 持久的Queue
                 *  4.1.3 持久的Topic
             *  4.2 事务
             *   *  4.2.1 producer提交时的事务
                 *  4.2.2 示例代码
                 *  4.2.3 控制台效果
             *  4.3 签收Acknowledge
             *   *  4.3.1 非事务
                 *   *  4.3.1.1 代码
                     *  4.3.1.2 控制台效果
                 *  4.3.2 事务
                 *   *  4.3.2.1 代码
                     *  4.3.2.2 控制台效果
                 *  4.3.3 签收和事务关系

# ActiveMQ #

*@Author:hanguixian*

*@Email:hn\_hanguixian@163.com*

## 四 JMS规范与落地产品 ##

### 1 是什么 ###

#### 1.1 JavaEE ####

*  JavaEE是一套使用Java进行企业级应用开发的大家一致遵循的13个核心规范工业标准。JavaEE平台提供了一个基于组件的方法来加快设计、开发、装配及部署企业应用程序
    
     *  JDBC(Java Database) 数据库连接
     *  JNDI(Java Naming and Directory Interfaces) Java的命名和目录接口
     *  EJB(Enterprise JavaBean)
     *  RMI(Remote Method Invoke) 远程方法调用Java IDL(Interface Description Language)/CORBA(Common Object Broker Architecture) 接口定义语言公用对象请求代理程序体系结构
     *  JSP(Java Server Pages)
     *  Servlet
     *  XML(Extensible Markup Language)可扩展白标记语言
     *  JMS(Java Message Service) Java 消息服务
     *  JTA(Java Transaction API) Java事务API
     *  JTS(Java Transaction Service) Java事务服务
     *  JavaMail
     *  JAF(JavaBean Activation Framework)

#### 1.2 Java Message Service(Java消息服务是JavaEE中的一个技术 ####

*  Java消息服务指的是两个应用程序之间进行异步通信的API,它为标准消息协议和消息服务提供了一组通用接口，包括创建、发送、读取消息等，用于支持JAVA应用程序开发。在JavaEE中，当两个应用程序使用JMS进行通信时，它们之间并不是直接相连的，而是通过一个共同的消息收发服务组件关联起来以达到解耦、异步、削峰的效果。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70]

### 2 MQ中间件的其他落地产品 ###

*  Kafka、RabbitMQ、RocketMQ、ActiveMQ …

### 3 JMS的组成结构和特点 ###

#### 3.1 JMS Provider ####

*  实现JMs接口和规范的消息中间件，也就是我们的MQ服务器

#### 3.2 JMS producer ####

*  消息生产者，创建和发送JMS消息的客户端应用

#### 3.3 JMS consumer ####

*  消息消费者，接收和处理JMS消息的客户端应用

#### 3.4 JMS message ####

##### 3.4.1 消息头 #####

*  JMSDestination：消息发送的目的地，主要是指Queue和Topic
 *  JMSDeliveryMode：
    
     *  持久模式和非持久模式。
     *  一条持久性的消息:应该被传送“一次仅仅一次”，这就意味者如果JMS提供者出现故障，该消息并不会丢失，它会在服务器恢复之后再次传递。
     *  一条非持久的消息:最多会传送一次，这意味这服务器出现故障，该消息将永远丢失。
 *  JMSExpiration
    
     *  可以设置消息在一定时间后过期，默认是永不过期
     *  消息过期时间，等于Destination的send方法中的timeToLive值加上发送时刻的GMT时间值。
     *  如果timeToLive值等于零，则JMSExpiration 被设为零，表示该消息永不过期。
     *  如果发送后，在消息过期时间之后消息还没有被发送到目的地，则该消息被清除。
 *  JMSPriority
    
     *  消息优先级，从0-9十个级别，0到4是普通消息,5到9是加急消息。
     *  JMS不要求MQ严格按照这十个优先级发送消息，但必须保证加急消息要先于普通消息到达。默认是4级。
 *  JMSMessageID
    
     *  唯一识别每个消息的标识由MQ产生

##### 3.4.2 消息体 #####

*  封装具体的消息数据
 *  5种消息格式
    
     *  TextMessage：普通字符串消息， 包含一个string
     *  MapMessage：一个Map类型的消息，key为string类型，而值为Java的基本类型
     *  BytesMessage：二进制数组消息，包含一个byte\[\]
     *  StreamMessage：Java数据流消息，用标准流操作来顺序的填充和读取。
     *  ObjectMessage：对象消息，包含一个可序列化的Java对象
 *  发送和接受的消息体类型必须一致对应

##### 3.4.3 消息属性 #####

*  如果需要除消息头字段以外的值，那么可以使用消息属性
 *  识别/去重/重点标注等操作非常有用的方
 *  是什么
    
     *  他们是以属性名和属性值对的形式制定的。可以将属性视为消息头的扩展，属性指定一些消息头没有包括的附加信息，比如可以在属性里指定消息选择器。
     *  消息的属性就像可以分配给一条消息的附加消息头一样。 它们允许开发者添加有关消息的不透明附加信息。它们还用于暴露消息选择器在消息过滤时使用的数据。

TextMessage message = session.createTextMessage();
    message.setText(text);
    message. setStringProperty("username'","z3"); /自定义属性

### 4 JMS的可靠性 ###

#### 4.1 持久化PERSISTENT ####

##### 4.1.1 参数设置说明 #####

*  持久
    
     *  `messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT)`
     *  当服务器宕机，消息不存在
 *  非持久
    
     *  `messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT)`
     *  当服务器宕机，消息依然存在
 *  默认是持久还是非持久？

##### 4.1.2 持久的Queue #####

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

/** 生产者新增：messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT) 消费者不变，保持queue name 一致 */
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    public class ReliabilityQueueProvider { 
    
        private static final String url = "tcp://xxx:61616";
    
        public static void main(String[] args) throws JMSException { 
    
            //1.创建连接工厂，按照给定的URL地址，使用默认的用户和密码
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
            //2.通过连接工厂获取connection并访问
            Connection connection = connectionFactory.createConnection();
            connection.start();
            //3.创建会话session
            //两个参数，第一参数：事务，第二个参数：签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //4.创建目的地，具体是队列Queue还是主题topic
            Queue queue01 = session.createQueue("queueReliability");
            //5.创建消息的生产者
            MessageProducer messageProducer = session.createProducer(queue01);
            //消息持久化PERSISTENT、非持久化NON_PERSISTENT
            messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
            //6. 通过消息生产者生产消息到MQ
            for (int i = 0; i < 3; i++) { 
                //6.1 创建消息
                //文本消息
                TextMessage textMessage = session.createTextMessage("hello activeMQ--queueReliability msg " + i);
                //6.2 发送到MQ
                textMessage.acknowledge();
                messageProducer.send(textMessage);
            }
    
            //7.关闭资源
            messageProducer.close();
            session.close();
            connection.close();
    
            System.out.println("消息发送成功************");
        }
    }

*  非持久下控制台变化：生产消息–>服务器宕机–>重启

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 1]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 2]

*  设置设置队列消息持久

//修该非持久代码段：messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);为-->
    messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);

*  持久下控制台：生产消息–>服务器宕机–>重启 控制台变化

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 3]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 4]

*  不写该段代码，队列默认为持久化。

##### 4.1.3 持久的Topic #####

*  类似微信公众号订阅发布
 *  持久化主题topic生产者代码

import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    public class ReliabilityTopicProvider { 
    
        private static final String url = "tcp://xxx:61616";
    
        public static void main(String[] args) throws JMSException { 
    
            //1.创建连接工厂，按照给定的URL地址，使用默认的用户和密码
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
            //2.通过连接工厂获取connection
            Connection connection = connectionFactory.createConnection();
            //3.创建会话session
            //两个参数，第一参数：事务，第二个参数：签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //4.创建目的地，具体是队列Queue还是主题topic
            Topic topic01 = session.createTopic("topicReliability");
            //5.创建消息的生产者
            MessageProducer messageProducer = session.createProducer(topic01);
          	//持久化
            messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);
    
            connection.start();
    
            //6. 通过消息生产者生产消息到MQ
            for (int i = 0; i < 3; i++) { 
                //6.1 创建消息
                //文本消息
                TextMessage textMessage = session.createTextMessage("hello activeMQ--topic Reliability msg " + i);
                //6.2 发送到MQ
                messageProducer.send(textMessage);
            }
    
            //7.关闭资源
            messageProducer.close();
            session.close();
            connection.close();
    
            System.out.println("消息发送成功************");
        }
    }

*  持久化主题topic消费者代码

import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    import java.io.IOException;
    
    public class ReliabilityTopicConsumer { 
        private static final String url = "tcp://xxx:61616";
    
        public static void main(String[] args) throws JMSException, IOException { 
            //1.创建连接工厂，按照给定的URL地址，使用默认的用户和密码
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
            //2.通过连接工厂获取connection并访问
            Connection connection = connectionFactory.createConnection();
            connection.setClientID("hgx_topic01");
            //3.创建会话session
            //两个参数，第一参数：事务，第二个参数：签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //4.创建目的地，具体是队列Queue还是主题topic
            Topic topic01 = session.createTopic("topicReliability");
            TopicSubscriber topicSubscriber = session.createDurableSubscriber(topic01, "remark");
    
            connection.start();
    
            Message message = topicSubscriber.receive();
            while (message instanceof TextMessage) { 
                TextMessage textMessage = (TextMessage) message;
                System.out.println("接收到持久化topic --->" + textMessage.getText());
                message = topicSubscriber.receive();
            }
    
            //关闭资源
            session.close();
            connection.close();
        }
    }

*  控制台：启动订阅者（消费者）–>关闭订阅者–>启动生产者（发布者）–>启动订阅者–>关闭订阅者

消费者启动

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 5]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 6]

消费者关闭

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 7]

生产者启动

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 8]

生产者启动，消费者启动–>消费成功

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 9]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 10]

消费者关闭

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 11]

#### 4.2 事务 ####

*  事务偏生产者，签收偏消费者

##### 4.2.1 producer提交时的事务 #####

*  false
    
     *  只要执行send，消息就进入到队列中
     *  关闭事务，那第二个签收参数需要设置有效
 *  true
    
     *  先执行send，再执行commit，消息才被真正的提交到队列中
     *  消息需要批量发送，需要缓冲区处理

##### 4.2.2 示例代码 #####

*  生产者

import org.apache.activemq.ActiveMQConnectionFactory;
    import javax.jms.*;
    
    public class TransactionQueueProvider { 
    
        private static final String url = "tcp://xxx:61616";
    
        public static void main(String[] args) throws JMSException { 
    
            //1.创建连接工厂，按照给定的URL地址，使用默认的用户和密码
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
            //2.通过连接工厂获取connection并访问
            Connection connection = connectionFactory.createConnection();
            connection.start();
            //3.创建会话session
            //两个参数，第一参数：事务，第二个参数：签收
            //true 开启事务
            Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
            //4.创建目的地，具体是队列Queue还是主题topic
            Queue queue01 = session.createQueue("queueTransaction");
            //5.创建消息的生产者
            MessageProducer messageProducer = session.createProducer(queue01);
            //6. 通过消息生产者生产消息到MQ
            for (int i = 0; i < 3; i++) { 
                //6.1 创建消息
                //文本消息
                TextMessage textMessage = session.createTextMessage("hello activeMQ--transaction msg " + i);
                //6.2 发送到MQ
                messageProducer.send(textMessage);
            }
    
            //7.关闭资源
            messageProducer.close();
            //提交事务
            session.commit();
            session.close();
            connection.close();
    
            System.out.println("消息发送成功************");
        }
    }

*  消费者

import org.apache.activemq.ActiveMQConnectionFactory;
    import javax.jms.*;
    import java.io.IOException;
    
    public class TransactionQueueConsumer { 
        private static final String url = "tcp://xxx:61616";
    
        public static void main(String[] args) throws JMSException, IOException { 
            //1.创建连接工厂，按照给定的URL地址，使用默认的用户和密码
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
            //2.通过连接工厂获取connection并访问
            Connection connection = connectionFactory.createConnection();
            connection.start();
            //3.创建会话session
            //两个参数，第一参数：事务，第二个参数：签收
            //开启事务
            Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
            //4.创建目的地，具体是队列Queue还是主题topic
            Queue queue01 = session.createQueue("queueTransaction");
            //5.创建消费者
            MessageConsumer messageConsumer = session.createConsumer(queue01);
    
            //消费方式1：同步阻塞方式（receive）订阅者或接收者调用MessageConsumer的receive方法来接收，receive方法在接收到消息之前或超时之前将一直阻塞
            while (true) { 
                Message message = messageConsumer.receive(4000);
                if (message instanceof TextMessage) { 
                    TextMessage textMessage = (TextMessage) message;
                    System.out.println(textMessage.getText());
                } else { 
                    break;
                }
    
            }
    
            //关闭资源
            messageConsumer.close();
            //提交事务
            session.commit();
            session.close();
            connection.close();
        }
    
    }

##### 4.2.3 控制台效果 #####

*  生产者启动，开始事务，不commit，控制台中可以看到没有任何消息进入MQ

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 12]

*  生产者启动，开始事务，并commit，控制台中可以看到消息进入MQ  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 13]
 *  消费者启动，开始事务，不commit，可以看到消息没有出MQ，消费者端可以重复消费

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 14]

*  消费者启动，开始事务，并commit，可以看到消息消费成功

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 15]

*  idea控制台打印如下

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 16]

#### 4.3 签收Acknowledge ####

##### 4.3.1 非事务 #####

*  自动签收（默认）：`Session.AUTO_ACKNOWLEDGE`
 *  手动签收:`Session.CLIENT_ACKNOWLEDGE`,客户端调用acknowledge方法手动签收
 *  允许重复消息:`Session.DUPS_OK_ACKNOWLEDGE`

###### 4.3.1.1 代码 ######

*  生产者

import org.apache.activemq.ActiveMQConnectionFactory;
    import javax.jms.*;
    
    public class AcknowledgeNoTransactionQueueProvider { 
    
        private static final String url = "tcp://xxx:61616";
    
        public static void main(String[] args) throws JMSException { 
    
            //1.创建连接工厂，按照给定的URL地址，使用默认的用户和密码
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
            //2.通过连接工厂获取connection并访问
            Connection connection = connectionFactory.createConnection();
            connection.start();
            //3.创建会话session
            //两个参数，第一参数：事务，第二个参数：签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //4.创建目的地，具体是队列Queue还是主题topic
            Queue queue01 = session.createQueue("queueAcknowledgeNoTransaction");
            //5.创建消息的生产者
            MessageProducer messageProducer = session.createProducer(queue01);
            //6. 通过消息生产者生产消息到MQ
            for (int i = 0; i < 3; i++) { 
                //6.1 创建消息
                //文本消息
                TextMessage textMessage = session.createTextMessage("hello activeMQ-- AcknowledgeNoTransaction msg " + i);
                //6.2 发送到MQ
                messageProducer.send(textMessage);
            }
    
            //7.关闭资源
            messageProducer.close();
            session.close();
            connection.close();
    
            System.out.println("消息发送成功************");
        }
    }

*  消费者

package com.hgx.activemq.acknowledge.consumers;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    import java.io.IOException;
    
    public class AcknowledgeNoTransactionQueueConsumer { 
        private static final String url = "tcp://106.14.217.80:61616";
    
        public static void main(String[] args) throws JMSException, IOException { 
            //1.创建连接工厂，按照给定的URL地址，使用默认的用户和密码
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
            //2.通过连接工厂获取connection并访问
            Connection connection = connectionFactory.createConnection();
            connection.start();
            //3.创建会话session
            //两个参数，第一参数：事务，第二个参数：签收
            Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
            //4.创建目的地，具体是队列Queue还是主题topic
            Queue queue01 = session.createQueue("queueAcknowledgeNoTransaction");
            //5.创建消费者
            MessageConsumer messageConsumer = session.createConsumer(queue01);
    
            //消费方式1：同步阻塞方式（receive）订阅者或接收者调用MessageConsumer的receive方法来接收，receive方法在接收到消息之前或超时之前将一直阻塞
            while (true) { 
                Message message = messageConsumer.receive(2000);
                if (message instanceof TextMessage) { 
                    TextMessage textMessage = (TextMessage) message;
                    //手动签收
                    message.acknowledge();
                    System.out.println(textMessage.getText());
                } else { 
                    break;
                }
    
            }
    
            //关闭资源
            messageConsumer.close();
            session.close();
            connection.close();
        }
    
    }

###### 4.3.1.2 控制台效果 ######

*  生产者启动，消费者启动，消费者设置手动签收`Session.CLIENT_ACKNOWLEDGE`，不写 `message.acknowledge()`,即不显示签收，消息不出MQ，消费者端存在重复消费

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 17]

*  生产者启动，消费者启动，消费者设置手动签收`Session.CLIENT_ACKNOWLEDGE`，写 `message.acknowledge()`,即显示签收，消息消费成功

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 18]

##### 4.3.2 事务 #####

*  事务开启后，只有commit后才能将消息生产或消费

###### 4.3.2.1 代码 ######

*  消息生产者

import org.apache.activemq.ActiveMQConnectionFactory;
    import javax.jms.*;
    
    public class AcknowledgeTransactionQueueProvider { 
    
        private static final String url = "tcp://xxx:61616";
    
        public static void main(String[] args) throws JMSException { 
    
            //1.创建连接工厂，按照给定的URL地址，使用默认的用户和密码
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
            //2.通过连接工厂获取connection并访问
            Connection connection = connectionFactory.createConnection();
            connection.start();
            //3.创建会话session
            //两个参数，第一参数：事务，第二个参数：签收
            Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
            //4.创建目的地，具体是队列Queue还是主题topic
            Queue queue01 = session.createQueue("queueAcknowledgeTransaction");
            //5.创建消息的生产者
            MessageProducer messageProducer = session.createProducer(queue01);
            //6. 通过消息生产者生产消息到MQ
            for (int i = 0; i < 3; i++) { 
                //6.1 创建消息
                //文本消息
                TextMessage textMessage = session.createTextMessage("hello activeMQ-- AcknowledgeTransaction msg " + i);
                //6.2 发送到MQ
                messageProducer.send(textMessage);
            }
    
            //7.关闭资源
            messageProducer.close();
            session.commit();
            session.close();
            connection.close();
    
            System.out.println("消息发送成功************");
        }
    }

*  消息消费者

import org.apache.activemq.ActiveMQConnectionFactory;
    import javax.jms.*;
    import java.io.IOException;
    
    /** * 在事务开启的情况下，签收参数无所谓，事务提交成功则默认签收，如果不commit，则签收无效，消息会被重复消费到 */
    public class AcknowledgeTransactionQueueConsumer { 
        private static final String url = "tcp://xxx:61616";
    
        public static void main(String[] args) throws JMSException, IOException { 
            //1.创建连接工厂，按照给定的URL地址，使用默认的用户和密码
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
            //2.通过连接工厂获取connection并访问
            Connection connection = connectionFactory.createConnection();
            connection.start();
            //3.创建会话session
            //两个参数，第一参数：事务，第二个参数：签收
            Session session = connection.createSession(true, Session.CLIENT_ACKNOWLEDGE);
            //4.创建目的地，具体是队列Queue还是主题topic
            Queue queue01 = session.createQueue("queueAcknowledgeTransaction");
            //5.创建消费者
            MessageConsumer messageConsumer = session.createConsumer(queue01);
    
            //消费方式1：同步阻塞方式（receive）订阅者或接收者调用MessageConsumer的receive方法来接收，receive方法在接收到消息之前或超时之前将一直阻塞
            while (true) { 
                Message message = messageConsumer.receive(1000);
                if (message instanceof TextMessage) { 
                    TextMessage textMessage = (TextMessage) message;
                    //手动签收
                    //message.acknowledge();
                    System.out.println(textMessage.getText());
                } else { 
                    break;
                }
    
            }
    
            //关闭资源
            messageConsumer.close();
            session.commit();
            session.close();
            connection.close();
        }
    
    }

###### 4.3.2.2 控制台效果 ######

*  生产者开启，并开启事务，生产消息，commit；消费者开启，并开启事务，不管ack参数是自动签收还是手动签收，只要没有commit提交，消息不会出MQ

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 19]

*  生产者开启，并开启事务，生产消息，commit；消费者开启，并开启事务，不管ack参数是自动签收还是手动签收，commit提交，消息会被消费

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 20]

##### 4.3.3 签收和事务关系 #####

*  签收和事务关系
    
     *  在事务性会话中，当一个事务成功提交则消息被自动签收。如果事务回滚，则消息会被再次传送
     *  在非事务性会话中，消息何时被确认取决于创建会话时的应答模式（Acknowledgement mode）

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191106234642365.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20191106233408813.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20191106233447212.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20191106233534214.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20191106233607638.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20191106233637939.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/20191106233705433.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 7]: https://img-blog.csdnimg.cn/20191106233736345.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 8]: https://img-blog.csdnimg.cn/20191106233809748.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 9]: https://img-blog.csdnimg.cn/20191106233831841.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 10]: https://img-blog.csdnimg.cn/20191106233854941.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 11]: https://img-blog.csdnimg.cn/20191106234034957.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 12]: https://img-blog.csdnimg.cn/20191106234100175.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 13]: https://img-blog.csdnimg.cn/20191106234123252.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 14]: https://img-blog.csdnimg.cn/20191106234143456.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 15]: https://img-blog.csdnimg.cn/20191106234203712.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 16]: https://img-blog.csdnimg.cn/20191106234223401.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 17]: https://img-blog.csdnimg.cn/20191106234248606.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 18]: https://img-blog.csdnimg.cn/20191106234307492.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 19]: https://img-blog.csdnimg.cn/20191106234333657.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1_size_16_color_FFFFFF_t_70 20]: https://img-blog.csdnimg.cn/20191106234352984.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hneF9zdWl5dWVzdXN1,size_16,color_FFFFFF,t_70