ActiveMQ 基础操作和应用

爱被打了一巴掌 2022-03-11 10:35 595阅读 0赞

**-----------------------------ActiveMQ**

**Java消息服务：** 两个系统之间或者分布式系统之间的信息通信。 一般我们使用 dubbo框架开发之后，因为有很多个 项目模块，每一个项目模块 都是一个 独立的 java项目，这些java项目之间 如果想进行 【信息】通信，这时候就要使用 【消息中间件服务】，这个【消息中间件服务】就是 我们的 jms规范，我们的 jms规范的 具体实现就是 我们下面要讲的 ActiveMQ ，我们是使用 ActiveMQ来 实现 两个系统或者 分布式 系统之间的 信息互通的。

**-----------------到底是啥意思呢？**

假如 项目1 想 发送一个 数据（字符串或者对象） 给我们的 项目2 ，我们就必须 让 【项目1】 先将 需要 发送的 数据 给我们的 【中间件】，然后再由我们的中间件 来替我们给 【项目2】。 两个独立的项目，没有任何耦合关系的两个项目。可以使用这个中间件来传递发送信息。

**为什么 需要 这样一个东西呢 ？**

回想一下之前使用dubbo框架来开发我们的分布式项目的时候：

（1）生产者 将 服务 注册 到 我们的【zookeeper注册中心】。

（2）消费者 订阅 【zookeeper注册中心】中的服务 。

**（3）这样子 就 实现了 生产者提供的服务 消费者可以使用。**

**如果 我们的 【消费者】 和 【生产者】 想要进行 【信息通信】，那么我们就要 加入 一个 【ActiveMQ消息中间件】 ，来让我们的生产者和消费者两个独立 项目 进行信息通信 。**

**有哪些优点 ？**

发送者A ------发消息----》 【中介】《------获取消息----接受者B

（1）我们的发送者A，发送了消息就发送了，不需要等待 接受者接受完消息，发送完就完了，任务就完成了，然后就进行下一次的发送消息的任务。

（2）我们的 发送者A 和 接收者B 是独立的，互不影响的，没有任何耦合的。

**ActiveMQ 是 JMS 规范的 具体实现 。**

**\*\*\*\*\*\*\*\*需要配置java环境变量的软件有：tomcat，maven，zookeeper，ActiveMQ。**

**因为这些 软件 都是使用 java语言进行开发的。因为是java语言开发的，所以解压后就可以直接使用。**

![20190307203150456.png][]

**ActiveMQ后台管理网站：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FtZWlyX3lhbmc_size_16_color_FFFFFF_t_70][]

圆圈1 ： 是管理菜单。

圆圈2 ： 是管理菜单中最重要的两个【选项】，左边的是管理【点对点通信模式】，右边的是管理【注册订阅通信模式】。

在 Linux 中开启 我们的 ActiveMQ服务：

![20190307203219243.png][]

到这个目录下，输入启动ActiveMQ 和关闭ActiveMQ的命令。

**启动ActiveMQ的命令 ：**

![20190307203234508.png][]

下面的info是启动activemq之后的日志打印。

**关闭ActiveMQ的命令 ：**

![20190307203319988.png][]

下面的一些 也是 打印的日志信息，不是错误。

**Windows 中 使用我们的 ActiveMQ ：**

1.  我们在windows中安装 activeMQ也是解压之后就可以使用。
2.  我们在windows中启动ActiveMQ，先切换到 ActiveMQ的bin目录下，然后在地址栏（目录栏）中输入cmd，然后就可以从控制台中输入 activemq start 这个命令来启动我们的ActiveMQ.
3.  退出和关闭我们的 ActiveMQ 的常用方法：
    
    1.  我们最常用的方式就是：ctrl+c 然后 输入 y，这样的方式关闭。
    2.  输入 activemq stop 来关闭。

**【生产者】，【消费者】，【中间件服务】的交互模型：**

![20190307203436105.png][]

【点对点】

这种通信模式 是： 【基于队列】 和 【一对一】 的，也就是说，发送者（生产者）发出的一个信息，只能被一个 消费者 【接收到】。 如果 有 很多个 【消费者】注册到我们的 ActiveMQ 中间件服务器中的时候，也就是有多个【消费者】想要获取 我们的这个 【生产者】发出的信息的时候，这种情况下也只能有一个 【消费者】获取到这个信息，因为是一对一模式，其他的消费者是获取不到的。

【发布/订阅】

这种模式 是 【一对多】的，也就是： 有一个【生产者】发送一个信息，然后有【很多个消费者】需要接收我们的这个信息，这时候我们使用这种模式下，我们的activeMQ中间件将这个信息 推送给每一个【消费者】，也就是说 每一个消费者都能获取到这一个生产者发送过来的信息 。

**这两种模式的使用场景**

实际开发中，我们一定会遇见【多线程高并发】的情况。当我们涉及到 多线程的情况的时候 我们 选取 【点对点】模式化。那么这些个线程中，只有一个线程对应的消费者可以获取到这个信息，因为我们的【点对点】模式，涉及到了lock的知识，可以看一下我们存放 信息的目录：

当我们 设置 【ActiveMQ 信息持久化】的时候，我们的ActiveMQ是将接收到的信息 存放到 ：/usr/local/apache-activemq-5.15.8/data/kahadb 中的,如下图：

![20190307203459522.png][]

**红圈就是 我们 【点对点】模式下的 lock线程锁。**

如果我们设置【消息不持久化】，那么我们的 ActiveMQ中间层服务 会将我们的【消息】，**存放在【内存】中**。

如果我们选取的是【发布/订阅】模式：那么就不用多说了，所有的线程对应的消费者都会获取到 生产者发送的信息 。

**特别需要说明一下的是：**

JMS是规范：

因为JMS是规范，所以JMS API中都是定义的接口，没有任何的实体类。

**ActiveMQ 目前还没有实现JMS2.0规范，所以不能使用注解是开发，主要原因是：**

我们jdk到8.0了，所以我们jms的规范也就更新到了2.0，jms的注解式开发必须是2.0以上的版本才可以。但是我们的ActiveMQ目前还没有实现 JMS2.0规范，所以我们使用ActiveMQ消息中间件进行开发的时候不能使用注解式开发的。

**JMS 和 ActiveMQ 所使用到的 相关依赖（jar）：**

<groupId>javax.jms</groupId>

<artifactId>javax.jms-api</artifactId>

</dependency>

<groupId>org.apache.activemq</groupId>

<artifactId>activemq-client</artifactId>

</dependency>

**为什么JMS的相关依赖是2.0.1呢？**

因为在maven的中央仓库中，可以找到的 JMS最新最新的依赖就是2.0.1.

如下图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FtZWlyX3lhbmc_size_16_color_FFFFFF_t_70 1][]

***实际开发小例子：***

创建 一个 生产者 和 一个 消费者：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FtZWlyX3lhbmc_size_16_color_FFFFFF_t_70 2][]

两个项目中 都要添加 JMS 和ActiveMQ 相关的依赖：

<dependency>
        <groupId>javax.jms</groupId>
        <artifactId>javax.jms-api</artifactId>
        <version>2.0.1</version>
    </dependency>
    
    
    <dependency>
        <groupId>org.apache.activemq</groupId>
        <artifactId>activemq-client</artifactId>
        <version>5.15.7</version>
    </dependency>

**Idea中：让 大写字母小写/小写字母大写 的 快捷键。 Ctrl + Shift + U**

**当我们的依赖出现缺少或者冲突的时候，我们应该怎么解决呢？**

当我们添加的依赖中，有缺少自己所需要依赖的jar包，但是没有添加进去。如：

我们在实现activeMQ的时候需要添加：

<dependency>
        <groupId>org.apache.activemq</groupId>
        <artifactId>activemq-client</artifactId>
        <version>5.15.7</version>
    </dependency>

这一段依赖。

但是activemq-client 这个依赖 有又要依赖 slf4j-api-1.7.25 这个依赖，但是我们在pom文件中添加进去 activeMQ-client这个依赖之后，我们的maven并没有自动帮我们导入activemq-client依赖所要依赖的slf4j-api-1.7.25，所以在运行的时候就会报出一段红色的错误：

![2019030720380374.png][]

这一段错误就是缺少slf4j-api-1.7.25 这个依赖所导致的，但是程序缺少这个依赖仍然可以正常执行，所以我们处理这种情况，第一种方法是先尝试 将这个依赖过滤掉：

过滤的方法如下：

<dependency>
        <groupId>org.apache.activemq</groupId>
        <artifactId>activemq-client</artifactId>
        <version>5.15.7</version>
        <exclusions>
            <exclusion>
                <groupId>org.slf4j</groupId>
                <artifactId>slf4j-api</artifactId>
            </exclusion>
        </exclusions>
    </dependency>

就是使用<exclusions>这个标签 将我们的 slf4j这个依赖排除掉。

但是我们这样做之后，回报出来另外的错误：

![20190307203914381.png][]

这个错误的意思是： 缺少 slf4j这个jar包，找不到这个jar包。

所以说 这个 jar 我不能直接进行过滤，必须要有的。

所以我们 使用第二个方法来解决：

那就是在pom文件中添加一个 slf4j-api-1.7.25 这个依赖的实现包：

但是我们在pom文件中添加slf4j-api-1.7.25这个jar的依赖的时候我们可以发现，我们mvn仓库中没有这个 jar的 依赖，所以我们只能找一个 实现了slf4j这个的实现包，如：添加下面这个依赖包之后就不会报错了。

<dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-simple</artifactId>
        <version>1.7.25</version>
    </dependency>

我们不能添加下面这个依赖包：因为我们添加下面这个依赖之后，我们启动main方法之后，还是会报之前那个错，所有我们需要添加上面的那么依赖包。

**【点对点的 发送 和 接收 代码】**

**生产者 发送信息 代码如下：**

package com.bjpowernode.activemq.queue;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    /**
     * ClassName:RunMain
     * Package:com.bjpowernode.activemq.queue
     * Decription:
     *
     * @data:2019/2/18 17:10
     * @author:yangZongJian
     */
    public class RunMain {
    
        private static final String BROKER_URL = "tcp://192.168.242.128:61616";
    
        private static final String DESTINATION_NAME = "myQueue";
    
    
        public static void main(String[] args) {
            sendMessage();
        }
    
        /**
         * 使用【点对点】的模式，发送【信息】。
         */
        public static void sendMessage(){
    
            ConnectionFactory connectionFactory = null ;
            Connection connection = null ;
            Session session = null ;
            MessageProducer producer = null ;
            Destination destination = null ;
            TextMessage textMessage = null ;
    
            try {
    
                //创建一个 【连接工厂】
                connectionFactory = new ActiveMQConnectionFactory(BROKER_URL);
                //通过 连接工厂 创建一个 【连接通道Connection】 。
                connection = connectionFactory.createConnection();
                //创建换一个会话session
                //第一个参数: 是否开启事务。Boolean.FALSE：不开启事务。Boolean.TRUE：开启事务。
                //第二个参数: 规定 消息确认 机制。Session.AUTO_ACKNOWLEDGE 遇到
                session = connection.createSession(Boolean.FALSE,Session.AUTO_ACKNOWLEDGE);
                //有了这个session之后，我们就可以创建 一个【消息】，然后向【消息】中添加：消息的内容，目的地，生产者，消费者。
                //创建一个 文本内容的 消息 。
                textMessage = session.createTextMessage("文本内容的信息......");//创建一个文本内容的 消息。
                //创建 消息的目的地。
                destination = session.createQueue(DESTINATION_NAME);
                //创建一个 生产者
                producer = session.createProducer(destination);
                //让生产者 发送 信息 。
                producer.send(textMessage); //发消息是没有返回值的，所以是非阻塞的，发完就完了，不需要等待任何结果的返回。void send()发送消息的方法。
    
            } catch (JMSException e) {
                e.printStackTrace();
            }finally {
                if(null != producer){
                    try {
                        producer.close();
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
                if(null != session){
                    try {
                        session.close();
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
                if(null != connection){
                    try {
                        connection.close();
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            }
    
        }
    
    }

**消费者端接收 消息的代码：**

package com.bjpowernode.activemq;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    /**
     * ClassName:RunMain
     * Package:com.bjpowernode.activemq
     * Decription:
     *
     * @data:2019/2/18 19:48
     * @author:yangZongJian
     */
    public class RunMain {
    
        public static final String BROKER_URL = "tcp://192.168.242.128:61616";
    
        public static final String DESTINATION = "myQueue" ;
    
        public static void main(String[] args) {
            receiveMessage();
        }
    
    
        public static void receiveMessage(){
    
            ConnectionFactory connectionFactory = null ;
            MessageConsumer consumer = null ;
            Connection connection = null ;
            Session session = null ;
            try {
                //创建一个连接工厂。
                connectionFactory = new ActiveMQConnectionFactory(BROKER_URL);
                //创建一个连接
                connection = connectionFactory.createConnection();
                /*
                        因为我们是【点对点】的模式，
                        所以需要我们的消费者自己去 获取 存放在ActiveMQ中的消息，
                        所以我们获取到了connection之后需要启动一下连接，
                        让这个连接连接到ActiveMQ上，这个步骤之后消费者才有。
                 */
                connection.start();
                //创建一次会话，也就是创建一个session。
                //第一个参数是： 是否开启事务、
                //第二次参数是：规定 信息确认 规则。
                session = connection.createSession(Boolean.FALSE,Session.AUTO_ACKNOWLEDGE);
                //创建一个目的地对象。
                Destination destination = session.createQueue(DESTINATION);
                //有了session，我们就可以创建消费者对象。
                consumer = session.createConsumer(destination);
                Message message = consumer.receive();
                if(message instanceof TextMessage){
                    String text = ((TextMessage) message).getText();
                    System.out.println(text);
                }
            } catch (JMSException e) {
                e.printStackTrace();
            }finally {
                if(null != consumer){
                    try {
                        consumer.close();
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
                if(null != session){
                    try {
                        session.close();
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
                if(null != connection){
                    try {
                        connection.close();
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            }
    
        }
    }

**【消费者】 和 【生产者】 代码之间的区别，不同之处如下：**

1.  生产者：

**创建session之后：**

可以使用session创建【消息对象】，【目的地对象】，【生产者对象】 这三个对象。

1.  消费者：

**创建session之后：**

可以使用session创建 【目的地对象】，【消费者对象】。

**创建connection连接对象之后：**

需要调用start()方法，开启一下连接。

**【发布订阅 模式 的 发送 和 接收 的代码】**

**【发布订阅模式】 需要注意的是** ： 发布订阅模式 是 一种 广播模式，也就是 我们的activeMQ就是广播台，当我们 运行 【生产者】之后，我们的生产者 发出一个消息，这个消息 被我们的 ActiveMQ 接收到 ，接收到之后，因为我们使用的是 发布订阅的模式，所以我们的 activeMQ 按照广播的模式进行 向 我们的【消费者】 推送和广播我们的 消息，如果我们的 【消费者 正在等待接收消息过程中】（因为我们使用的recevie这个方法，所以当我们的消费者没有获取到消息的时候，就会阻塞一直等待接收消息），我们的消费者就会接收到 activeMQ 广播过来的消息，如果我们的 【消费者没有处于等待】，那么消费者就错过了广播，也就是没听到广播信息，这样子就没有获取到信息。所以说 我们的 【发布订阅】的模式会有信息的丢失。 【发布订阅模式】 是强行将信息 推送给 消费者，它不管你有没有处于等待接收的状态，他只管推送，所以我们在接收消息的时候，需要先启动我们的消费者，让消费者处于接收消息等待的状态，然后我们 再启动 生产者，让生产者发出消息，然后 再让ActiveMQ广播出这条消息 。

**【点对点模式】：** 我们的点对点模式 就 不会有 信息的丢失，因为 我们的消费者 在点对点模式中是 自己主动去 自己需要的 信息。 也就是，我们的生产者 产生一条信息之后，这条信息就会被我们ActiveMQ保存起来（持久化），然后只要我们的 服务器一直启动着，我们的 消费者可以 第一天来取，也可以可以第二天来取，因为都不会丢失，之所以这样是因为，我们的ActiveMQ默认是 将为取过的信息 持久化的，所以没有被消费者取过的信息会被 ActiveMQ保存在 磁盘的 data目录下。

【发布订阅模式】 是 广播模式，强行将信息推送给 【消费者】。

【点对点模式】 是 我们 【消费者】主动去 ActiveMQ中，获取信息。

**Topic主题下 生产者代码 和 消费者代码：**

就是将 【点对点】 模式下的代码中的：

生产者代码和消费者代码中：

创建目的地对象的代码修改为：

CreateTopic()方法就可以了

**【点对点】 和 【发布/订阅】的 比较 :**

（1）

【点对点】是【一对一】的模式。

【发布/订阅】是【一对多】的模式。

（2）

【点对点】模式 ：可以保证每条数据（信息）都可以被接收到

【发布/订阅】模式：不能保证每条数据都可以接受到。因为发布订阅给消费者信息的 模式是 广播模式，只有在听广播的消费者才可以获取到 信息数据。

1.  【点对点】： 生产者将 信息 传递给我们的 ActiveMQ，然后我们的ActiveMQ会将这个信息数据保存起来，然后等待 我们的消费者自己来获取，注意了自己来获取，什么时候来取都行，都有这条数据，只要我们的ActiveMQ的服务器正常运行。

【发布订阅】：生产者 将信息 传递给我们的 ActiveMQ ，然后我们的 ActiveMQ会将这个信息保存起来，然后 也就是 当ActiveMQ 获取到这个数据信息的时候，我们的ActiveMQ 开始了 广播模式，也就是 这个时候 那个【消费者】正在监听我们的ActiveMQ 谁就能获取到这个数据信息。

1.  【点对点】：ActiveMQ 中的信息 已经确认被【一个】消费者获取完毕之后，我们的activemq就会将这个信息删除掉，当一个消费者获取到之后就将这个信息删除，所以其他的消费者获取不到的。

【发布订阅】： ActiveMQ 中的数据信息，被【所有的想要获取这个信息数据的消费者】获取后，注意了是所有的想要获取这个数据信息的消费者都获取到了之后，我们的ActiveMQ才会将这个数据信息删除掉。

**拉模式 与 推模式 ：**

**拉模式 其实就是 点对点模式：**

我们 生产者 产生一个 信息数据，然后这个信息数据就保存在我们的ActiveMQ消息中间件中， 这时候 【消费者】来主动获取这个信息数据，获取信息完毕确认之后，ActiveMQ 就会将这个信息数据删除。如果没有【消费者】主动过来获取这个信息数据，那么我们ActiveMQ就会一直将这个 信息数据存放起来，等待消费者的获取。

**推模式 其实就是 发布订阅 模式：**

我们的 生产者 产生一个 信息数据 ，然后这个信息数据 就保存在我们的ActiveMQ消息中间件中，当我们的ActiveMQ获取到这个信息数据之后，就开始了广播模式，向 【想要获取这个信息数据的所有的消费者】进行广播，这时候 谁在听广播谁就能获取到我们的这个 信息数据，谁没有监听我们的 activeMQ就等于没有在听广播，就获取不到信息数据。

所以我们想让我们的消费者获取到信息数据，我们就必须让我们的消费者先启动，等待获取activceMQ推送过来的广播消息，然后开启生产者，让生产者发出信息，然后通过ActiveMQ广播信息，强行推送给我们正在监听的消费者。

**当我们 传递的信息为 Object类型 的时候：**

我们在传递一个Object的信息的时候，如果我们不设置一下可信任的Object类，那么消费者在启动的时候会报错的：会被报出 object这个类是不可信任的。主要异常内容如下：javax.jms.JMSException: Failed to build body from content. Serializable class not available to broker. Reason: java.lang.ClassNotFoundException: Forbidden class com.bjpowernode.model.User! This class is not trusted to be serialized as ObjectMessage payload. Please take a look at http://activemq.apache.org/objectmessage.html for more information on how to configure trusted classes.

**我们需要传递的这个类应该做哪些处理？**

将这个 类 进行 序列化。如果不设置无法在网络中进行传输。

将这个 类 设置为 可信任的。如果不设置无法被消费者解析获取。

1.      connectionFactory = new ActiveMQConnectionFactory(BROKER_URL);
                List<String> package_url = new ArrayList<String>();
                package_url.add("com.bjpowernode.model");
                //所有可以信任的Object类都可以一起添加进：list集合中。
                connectionFactory.setTrustedPackages(package_url);//这个是手动指定所信任的包。
                connection = connectionFactory.createConnection();
                connection.start();//写消费者 必须有的，没有为什么。
    
    **上面是第一种添加可以信任Object类的方法。比较麻烦。**
    
    **第二种 方法是： 给所有的Object类 统一设置为可信任 。**

connectionFactory = new ActiveMQConnectionFactory(BROKER_URL);
    //给所有的 Object 类 统一设置为 可以信任。
    connectionFactory.setTrustAllPackages(true);
    connection = connectionFactory.createConnection();
    connection.start();//写消费者 必须有的，没有为什么。

实际开发中不会 直接接收 一个 Object类型的对象的，我们一般将这个对象转换成json格式的String字符串来进行接收，也就是最后我们还是将一个Object类型的信息转换成了String字符串类型的文本信息，然后从后台接收这个json字符串文本信息。

使用到的技术是fastjson：
    String userJsonString = JSONObject.toJSONString(user);

### 过滤消息的方法： ###

**MapMessage 映射消息：**携带一组键值对的数据作为有效负载的消息，有效数据值必须是Java原始数据类型（或者它们的包装类）及String。即：byte, short, int, long, float, double, char, boolean, String。

说白就是 原先的 Map集合，使用键值对的方式存放信息。

**使用方式如下：**

**生产者 创建 map映射消息的方法：**

MapMessage mapMessage = session.createMapMessage();
    mapMessage.setBoolean("booleanKey",true);
    mapMessage.setDouble("doubleKey",3.14);
    mapMessage.setString("stringKey","张三");

**消费者 接收信息后，判断是否是 map映射消息的方法：**

if(message instanceof MapMessage){
        double doubleValue = ((MapMessage) message).getDouble("doubleKey");
        String stringValue= ((MapMessage) message).getString("stringKey");
        System.out.println(doubleValue);
        System.out.println(stringValue);
    }

**通常来讲，使用 instanceof 就是判断一个实例是否属于某种类型****。**

**ActiveMQ事务消息和非事务消息** **测试模式【点对点模式】**

**生产者的session：**

1.  添加事务： 将session的第一个参数设置为：Boolean.true。这样子就开启了事务机制。生产者调用send方法之后，我们立即查看 后台管理网站，发现消息队列中没有增加消息，这是为什么呢？ 这是因为 我们在 生产者中开启事务之后，手动的commit一下才会将消息提交到 ActiveMQ当中。使用 session.commit() 手动 commit。
2.  没有添加事务： 我们调用send方法之后就自动将 消息 提交到 我们的 ActiveMQ当中了。

**消费者的session：**

1.  添加事务： 如果我们不手动的session.commit（）提交一下，虽然我们消费者获取到了信息，但是 消息获取没有确认，所以消息会永久的保存在 ActiveMQ当中了，不会被删除，因为消息获取确认没有提交。 我们只有调用session.commit()来手动提交一下 消息获取确认，确认被ActiveMQ接收后就会将这个消息删除掉。 消息就等于是出队了。
2.  不添加事务： 如果不添加事务的话，当我们调动receive方法之后就会根据我们的配置的 消息确认方式 来进行消息确认的提交，消息确认提交的方式的配置就在 session的第二个参数。

**需要注意的是： 生产者 和 消费者 可以同时配置 事务，也可以不同时 配置事务机制。**

**我们可以在完成业务逻辑之后 提交 消息获取确认，也可以在业务逻辑之前 提交 消息获取确认，因为 这个消息获取确认的 提交 只是为了让 ActiveMQ 知道我们消费者已经获取到了 消息，好让我们的ActiveMQ 删除 消息。 我们的 消息确认的提交和业务逻辑的执行时 相互独立的。**

**当我们 给 消费者 设置了 事务机制之后，不管第二个参数 采用任何的 确认机制，我们的消息都会在 事务机制的最后 手动session.commit（）提交的时候 自动的进行消息确认的提交。就算我们配置第二个参数为 不提交消息确认，我们在手动commit的时候就已经自动的将 消息确认提交了。**

session = connection.createSession(Boolean.true,Session.AUTO_ACKNOWLEDGE);
    后面的Session.AUTO_ACKNOWLEDGE第二个参数配置什么都已经无所谓了，因为最后 commit的时候都会 自动提交 消息确认。 
    
    但是我们事务机制的标准session写法:
    session = connection.createSession(Boolean.true,Session.SESSION_TRANSACTED);
    虽然后面第二个参数写什么都无所谓了，但是这样写比较规范。

**消息的过滤：**

**生产者所需要的操作 ：**

TextMessage textMessage = session.createTextMessage(userString);
    //实现 消息 的 【过滤筛选】。 给我们的 消息 添加一个 【-标记-】 ，这个标记可以是 int,String,double。
    textMessage.setStringProperty("name","张三");
    textMessage.setIntProperty("age",18);

实现消息的 【过滤筛选】 的功能，必须在 【生产者】产生【消息】的时候 为消息添加上标记，这个标记是 我们【消费者】进行筛选过滤操作所需要用的。

**消费者所需要的操作：**

使用session 创建消费者对象的时候，将【过滤筛选条件】当作参数传递给我们消费者对象。到时候我们消费者从 ActiveMQ 中获取到 消息的时候 就可以根据 这个过滤条件进行筛选了 。

session = connection.createSession(Boolean.TRUE,Session.SESSION_TRANSACTED);
    //创建目的地
    Destination destination = session.createQueue(DESTINATION);
    //创建消费者
    consumer = session.createConsumer(destination,"name='张三' and age=18");

这个筛选条件的格式为: SQL语句中 where 后面的 条件格式 一样的 。 对比着sql语句中的where 来进行 书写。

**事务机制：**

**生产者和消费者都可以添加 事务机制，添加的格式都是一样的，标准格式如下****：**

session = connection.createSession(Boolean.TRUE,Session.SESSION_TRANSACTED);

在 【生产者】 项目中 添加 事务机制：

在【生产者】中添加了事务机制之后，生产者 发出消息之后，如果不手动session.commit()一下，那么消息就不会被提交到ActiveMQ当中，所以当生产者在send（）完毕消息之后，需要手动session.commit（）提交一次。

在【消费者】 项目中 添加 事务机制：

在【消费者】中添加添加了事务机制之后， 如果在onMessage或者receive方法之后没有 ，手动的session.commit()一下，虽然我们的 消费者 可以正常的从ActiveMQ 中获取到消息，但是因为没有commit，所以消息的确认没有提交，所以ActiveMQ中的这个条消息，在被消费者获取到之后没有被删除掉，被持久的保存在服务器磁盘中了。

**同步接收 和 异步接收**

【异步接收】：

（1）异步接收 其实 就是 两个线程。

（2） 主线程 从上往下执行，遇到

consumer.setMessageListener(new MessageListener() {}

这个方法 就 又分出去 一条线程 ，这个线程 就是 我们的 监听线程 。

1.  主线程 从上向下执行完毕之后，不能关闭connection，因为如果主线程将 connection连接关闭了，我们的 监听线程就不能正常执行了。
2.  这是两个线程 公用一个 连接通道connection。
3.  主方法执行完了，我们的监听线程还没有执行完。监听线程会一直对ActiveMQ 进行监听。

**异步请求可以实现对ActiveMQ的全天候的监听。**

final Session finalSession = session;
     consumer.setMessageListener(new MessageListener() {
        public void onMessage(Message message) {
            if(message instanceof TextMessage){
                String text = null;
                try {
                    text = ((TextMessage) message).getText();
                    finalSession.commit();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
                System.out.println(text);
            }
        }
    });

【同步接收】：

1.  只有一条线程，主线程，主线程从上往下执行，当 消费者 调用了 recevie()方法之后，调用完成之后 获取到消息，主线程接着往下执行，执行到最后，没有可以执行的了，线程销毁了。

**不能像异步请求那样自动的全天候的对ActiveMQ进行监听。**

**\*\*\*\*\*\*\*\*如果我们 想 让 同步接收 像 异步接收 那样 实时的对 ActiveMQ 进行 监听 ，我们就需要给 我们的 同步接收加一个 死循环：**

//创建消费者
    consumer = session.createConsumer(destination,"name='张三' and age=18");
    //获取消息,我们使用【异步接受 消息 的方式】
    
    while(true) {
        Message message = consumer.receive();
        if(message instanceof TextMessage){
            String text = ((TextMessage) message).getText();
            System.out.println(text);
        }
        session.commit();
    }

（1）Recevie() 方法 其实就是 单线程（主线程）的方法。

（2）Consumer.setMessageListener(new MessageLister()\{

Public void onMessage(Message message)\{

业务逻辑代码；

\});

主线程从上往下执行，当遇到 consumer.setMessageListener()这个方法的时候，就会为我们这个监听器 分配 一个 单独的 线程 ，用来监听 ActiveMQ，我们主线程继续往下执行，但是我们的 监听线程继续监听ActiveMQ 。

**特别需要注意的是：** **（1）和（2）不能在一起使用，也就是要么你使用第一种接收请求的方法，也就是使用同步接受方式，要么你就使用第二个接收请求的方法，也就是异步接收方式，你不能即使用 异步接收 和 同步接收 ，因为这样 会 报出 异常的 。**

**ActiveMQ 消息持久化**

ActiveMQ 默认是 进行 消息持久化的，但是 我们的消息 最好不要持久化，因为数据持久化操作是我们的 redis 和 mysql 的 功能，我们的ActiveMQ的功能就是一个中介的作用，帮助两个独立，互不干涉，不相互耦合的项目 进行 消息通信，起到了转发消息的作用。

**什么情况下 我们的 消息 会被永久的保存在了 ActiveMQ的data下的kahadb 目录下了？**

【点对点模式】： 我们的点对点模式 是 【异步提交确认】，也是【拉模式】，所以我们的消费者 是自己 主动去 ActiveMQ 中获取的消息，如果消费者一直不取，我们的ActiveMQ就会一直帮我们的 消费者存放着，一直到消费者获取了消息之后，才会删除这条消息，所以我们的 当我们 消费者 down机了，那么消息就永久的存放在了ActiveMQ 。

【发布/订阅模式】：发布订阅模式 是 一种 【广播模式】，也是 【推模式】，所以 消息是 我们的 ActiveMQ 主动强行推送广播给 所有需要接收的消费者的。 当我生产者 发出这条消息之后，这条消息被ActiveMQ接收到了，然后ActiveMQ会立即进行广播，在场的所有的消费者，谁正在监听ActiveMQ，谁就能听到广播，谁就能获取这条信息，而且ActiveMQ 就会广播一次，不会广播第二次的。 当我们的消费者接收到了 消息之后，没有提交信息获取确认，那么我们的 ActiveMQ就无法对消息进行删除，那么这个信息就永久的存放在了Active的kahadb目录下了。

**信息 持久化 存放的目录为：**

![2019030720532330.png][]

1.  我们的 ActiveMQ 默认是将消息持久化的。这样消息持久化的好处是：当ActiveMQ所在的服务器关闭了，重启之后， 之前的消息还在ActiveMQ 中，消费者仍然可以进行获取 。
2.  如果【消息不持久化】的话，这些 ActiveMQ接收到的消息 都会被保存在 【内存】中，这样子有一个很大的缺陷，消息很容易丢失，我们的服务器突然停机了，保存在内存中的 消息 全部丢失了。 实现消息不持久化的代码如下：

producer = session.createProducer(destination);
    //设置 producer生产者 传送的 消息 不持久化。
    producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
    producer.send(textMessage);
    //提交一下事务，将我们的 消息 提交到 ActiveMQ 中 。
    session.commit();

**提交消息之前：**

![2019030720541362.png][]

**提交消息之后：**

![20190307205422511.png][]

**当我们将ActiveMQ所在的服务器关闭之后再打开，然后登录到ActiveMQ管理界面后:**

![20190307205433349.png][]

因为 我们设置的 producer生产者 上传的消息是不持久化的，所以这些上传的消息都存放在了 【内存】当中了，所以我们关闭服务器之后，内存被清空，这些消息丢失了，所以出现了 消息减少的情况。

[20190307203150456.png]: /images/20220311/06d11b74da304775bff8e64811fda5bf.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FtZWlyX3lhbmc_size_16_color_FFFFFF_t_70]: /images/20220311/d05e6c5fa75f433da9cd10d4d5bb3080.png
[20190307203219243.png]: /images/20220311/f7b56d23d37144bdac4a4c316ba33ae3.png
[20190307203234508.png]: /images/20220311/ed27f527a1914a98a94cb59280d0557e.png
[20190307203319988.png]: /images/20220311/7c807b89f0884dcabf8d75640b8e92ff.png
[20190307203436105.png]: /images/20220311/1a5b8a2bc3054208904ffe1bc7524b38.png
[20190307203459522.png]: /images/20220311/5fdc38491b7c419481e619c6f85b8299.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FtZWlyX3lhbmc_size_16_color_FFFFFF_t_70 1]: /images/20220311/9bfd6e8a345346d7a65c8a59bcb7ddb5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FtZWlyX3lhbmc_size_16_color_FFFFFF_t_70 2]: /images/20220311/b9b06195086447f5b0011b08a13340d6.png
[2019030720380374.png]: /images/20220311/13b4d1e711484a26af64a4173741d28e.png
[20190307203914381.png]: /images/20220311/afde3b402a8c402b83e461914487f34b.png
[2019030720532330.png]: /images/20220311/5fb6d6b59fc34a29ae69897047666610.png
[2019030720541362.png]: /images/20220311/8de989a504e54f3a9974bbf07b77718a.png
[20190307205422511.png]: /images/20220311/6eabee2d277e4fa8942841e0c55ed387.png
[20190307205433349.png]: /images/20220311/bacdec73e94e423e98472fb9ab55fc32.png