消息中间件-ACTIVEMQ-10：ActiveMQ的高级特性

我不是女神ヾ 2023-10-05 09:55 67阅读 0赞

**目录**

1：异步投递

1.1：异步投递的含义

1.2：开启异步投递：

1.3： 确保异步发送成功：

2： 延时投递和定时投递

2.1：在activemq.xml中配置schedulerSupport属性为true，标识mq服务器开启定时和延时

2.2：代码层面的封装辅助消息类型schedulerMessage

1：生产者

2：消费者不用动

3：结果

3：消息重发机制

3.1：那些情况会引起消息重发

3.2：消息重发时间间隔和重发次数

3.3：有毒消息Poison ACK

3.4：消息重发的属性

3.5：消息重发的案例

3.6：消费者修改重发机制

3.7：spring整合后如何修改重发机制

4：死信队列

1：简介：

2：作用

3：自动删除过期消息

4：存放非持久的消息到死信队列

5：重复消息和幂等调用

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

# 1：异步投递 #

## 1.1：异步投递的含义 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70][]

## 1.2：开启异步投递： ##

## ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 1][]1.3： 确保异步发送成功： ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 2][]

package com.wkl.queuq;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.apache.activemq.ActiveMQMessageProducer;
    import org.apache.activemq.AsyncCallback;
    import org.springframework.beans.factory.annotation.Qualifier;
    
    import javax.jms.*;
    import java.util.UUID;
    
    /**
     * Description:第一次连接ActvieMq的服务
     * Date:       2020/9/3 - 下午 1:51
     * author:     wangkanglu
     * version:    V1.0
     */
    public class Produce {
    
        /*账号密码如果都是默认的admin，可以不用穿；直传url*/
        private static  final String USERNAME = "admin";
        private static  final String PASSWORD = "admin";
        /*这个url以tcp协议开头，java程序访问的是61616端口*/
        private static  final String ACTIVE_URL = "tcp://192.168.43.122:61616";
        private static  final String Queue_NAME = "queue01";
    
        public static void main(String[] args) throws JMSException {
            //1:创建连接工厂，按照给定的账号，密码，url地址来链接；
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, ACTIVE_URL);
            activeMQConnectionFactory.setUseAsyncSend(true);
            //2：通过链接工厂获得connection并启动
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
            //3：通过connection创建会话,参数分别是事务和签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //4：创建目的地（队列还是主题）
            Queue queue = session.createQueue(Queue_NAME);
            //5：创建消息生产者，
            ActiveMQMessageProducer producer = (ActiveMQMessageProducer) session.createProducer(queue);
            /*---------开启持久化-----------*/
            producer.setDeliveryMode(DeliveryMode.PERSISTENT);
    
            //6：通过消息生产者producer产生3条消息发到mq的队列中
            for (int i = 1; i <=5 ; i++) {
                //7：创建消息
                TextMessage textMessage = session.createTextMessage("msg--" + i);
                /*在这里设置发送消息的回调*/
                textMessage.setJMSMessageID(UUID.randomUUID().toString());
                String jmsMessageID = textMessage.getJMSMessageID();
    
                //8：通过perducer发送给mq
                producer.send(textMessage, new AsyncCallback() {
                    @Override
                    public void onSuccess() {
                        System.out.println("成功："+jmsMessageID);
                    }
    
                    @Override
                    public void onException(JMSException e) {
                        System.out.println("失败："+jmsMessageID);
                    }
                });
            }
    
            //9：关闭资源；顺着申请，倒着关闭
            producer.close();
            session.close();
            connection.close();
    
            System.out.println("---end---");
    
        }
    }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 3][]

# 2： 延时投递和定时投递 #

通过配置属性可以实现延时和定时投递信息，属性如下

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 4][]

## 2.1：在activemq.xml中配置schedulerSupport属性为true，标识mq服务器开启定时和延时 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 5][]

## 2.2：代码层面的封装辅助消息类型schedulerMessage ##

### 1：生产者 ###

package com.wkl.queuq;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.apache.activemq.ActiveMQMessageProducer;
    import org.apache.activemq.AsyncCallback;
    import org.apache.activemq.ScheduledMessage;
    import org.springframework.beans.factory.annotation.Qualifier;
    
    import javax.jms.*;
    import java.util.UUID;
    
    /**
     * Description:第一次连接ActvieMq的服务
     * Date:       2020/9/3 - 下午 1:51
     * author:     wangkanglu
     * version:    V1.0
     */
    public class Produce {
    
        /*账号密码如果都是默认的admin，可以不用穿；直传url*/
        private static  final String USERNAME = "admin";
        private static  final String PASSWORD = "admin";
        /*这个url以tcp协议开头，java程序访问的是61616端口*/
        private static  final String ACTIVE_URL = "tcp://192.168.43.122:61616";
        private static  final String Queue_NAME = "queue01";
    
        public static void main(String[] args) throws JMSException {
            //1:创建连接工厂，按照给定的账号，密码，url地址来链接；
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, ACTIVE_URL);
            activeMQConnectionFactory.setUseAsyncSend(true);
            //2：通过链接工厂获得connection并启动
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
            //3：通过connection创建会话,参数分别是事务和签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //4：创建目的地（队列还是主题）
            Queue queue = session.createQueue(Queue_NAME);
            //5：创建消息生产者，
            ActiveMQMessageProducer producer = (ActiveMQMessageProducer) session.createProducer(queue);
            /*---------开启持久化-----------*/
            producer.setDeliveryMode(DeliveryMode.PERSISTENT);
    
            //设置延迟时间
            long delay = 3*1000;
            long period = 4*1000;
            int repeat = 5;
    
            //6：通过消息生产者producer产生3条消息发到mq的队列中
            for (int i = 1; i <=5 ; i++) {
                //7：创建消息
                TextMessage textMessage = session.createTextMessage("msg--" + i);
                /*在这里设置发送消息的回调*/
                textMessage.setJMSMessageID(UUID.randomUUID().toString());
    
                //设置间隔发送的属性
                textMessage.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY,delay);
                textMessage.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_PERIOD,period);
                textMessage.setIntProperty(ScheduledMessage.AMQ_SCHEDULED_REPEAT,repeat);
    
                String jmsMessageID = textMessage.getJMSMessageID();
    
                //8：通过perducer发送给mq
                producer.send(textMessage, new AsyncCallback() {
                    @Override
                    public void onSuccess() {
                        System.out.println("成功："+jmsMessageID);
                    }
    
                    @Override
                    public void onException(JMSException e) {
                        System.out.println("失败："+jmsMessageID);
                    }
                });
            }
    
            //9：关闭资源；顺着申请，倒着关闭
            producer.close();
            session.close();
            connection.close();
    
            System.out.println("---end---");
    
        }
    }

### 2：消费者不用动 ###

### 3：结果 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 6][]

# 3：消息重发机制 #

## 3.1：那些情况会引起消息重发 ##

![20200911163024217.png][]

## 3.2：消息重发时间间隔和重发次数 ##

![20200911163034847.png][]

## 3.3：有毒消息Poison ACK ##

![20200911163118925.png][]

## 3.4：消息重发的属性 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 7][]

## 3.5：消息重发的案例 ##

消费者使用事务，但是读取完不提交

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 8][]

重复第6次后就不会再读到了；转入到死信队列；

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 9][]

## 3.6：消费者修改重发机制 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 10][]

## 3.7：spring整合后如何修改重发机制 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 11][]

# 4：死信队列 #

## 1：简介： ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 12][]

## 2：作用 ##

处理失败的消息

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 13][]

## 3：自动删除过期消息 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 14][]

## 4：存放非持久的消息到死信队列 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 15][]

# 5：重复消息和幂等调用 #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 16][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200911154206470.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/2020091115512519.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200911155414244.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200911160318999.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20200911161124959.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20200911161300844.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/20200911162553441.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2,size_16,color_FFFFFF,t_70
[20200911163024217.png]: https://img-blog.csdnimg.cn/20200911163024217.png
[20200911163034847.png]: https://img-blog.csdnimg.cn/20200911163034847.png
[20200911163118925.png]: https://img-blog.csdnimg.cn/20200911163118925.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 7]: https://img-blog.csdnimg.cn/20200911194447959.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 8]: https://img-blog.csdnimg.cn/20200911195257306.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 9]: https://img-blog.csdnimg.cn/20200911195441121.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 10]: https://img-blog.csdnimg.cn/20200911195910986.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 11]: https://img-blog.csdnimg.cn/20200911200035917.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 12]: https://img-blog.csdnimg.cn/20200914141248489.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 13]: https://img-blog.csdnimg.cn/20200914141410941.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 14]: https://img-blog.csdnimg.cn/20200914142151670.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 15]: https://img-blog.csdnimg.cn/20200914142355671.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2_size_16_color_FFFFFF_t_70 16]: https://img-blog.csdnimg.cn/20200914143409856.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjk0OTA2,size_16,color_FFFFFF,t_70