spring activeMQ 死信队列

Bertha 。 2022-07-18 00:18 207阅读 0赞

# 1:死信队列简介 #

DLQ-死信队列(Dead Letter Queue)用来保存处理失败或者过期的消息。  
出现以下情况时，消息会被redelivered  
 A transacted session is used and rollback() is called.

A transacted session is closed before commit is called.  
 A session is using CLIENT\_ACKNOWLEDGE and Session.recover() is called.  
当一个消息被redelivered超过maximumRedeliveries(缺省为6次，具体设置请参考后面的链接)次数时，会给broker发送一个"Poison ack"，这个消息被认为是a poison pill，这时broker会将这个消息发送到DLQ，以便后续处理。  
缺省的死信队列是ActiveMQ.DLQ，如果没有特别指定，死信都会被发送到这个队列。  
缺省持久消息过期，会被送到DLQ，非持久消息不会送到DLQ  
可以通过配置文件(activemq.xml)来调整死信发送策略。

### 注:本次测试使用的是第三种,使用自行确认接收并且在失败时session.recover()进行恢复. ###

# 2:activemq.xml配置 #

## 死信队列配置 ##

## 持久化配置(mysql) ##

# 3:RedeliveryPolicy重发策略设置 #

<bean id="activeMQConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://localhost:61636"></property>
        <property name="useAsyncSend" value="true"></property>
        <property name="redeliveryPolicy">
            <bean id="redeliveryPolicy" class="org.apache.activemq.RedeliveryPolicy">
                
                <property name="useExponentialBackOff" value="true"></property>
                
                <property name="maximumRedeliveries" value="5"></property>
                
                <property name="initialRedeliveryDelay" value="1000"></property>
                
                <property name="backOffMultiplier" value="2"></property>
                
                <property name="maximumRedeliveryDelay" value="1000"></property>
            </bean>
        </property>
    </bean>

# 4:send测试类 #

import org.junit.Test;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jms.core.JmsTemplate;
    
    /** * 类描述:sender测试类 * * @author fengyong * @version 1.0 * @since 1.0 * Created by fengyong on 16/8/3 下午7:45. */ public class ActiveMqSender extends BaseTest {
        @Autowired
        private JmsTemplate senderJmsTemplate;
    
        @Test
        public void activeMq(){
            for(int i = 1;i<=10;i++){
                senderJmsTemplate.convertAndSend("FirstQueue","我是第"+i+"个");
            }
            System.out.print("全部执行完毕!!!");
        }
    }

# 5:receiver测试类 #

import com.system.freemwork.amq.SimpleJmsTemplate;
    import org.apache.activemq.command.ActiveMQQueue;
    import org.junit.Test;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jms.core.JmsTemplate;
    
    import javax.jms.JMSException;
    import javax.jms.TextMessage;
    
    /** * 类描述:receiver测试类 * * @author fengyong * @version 1.0 * @since 1.0 * Created by fengyong on 16/8/8 下午5:28. */ public class ActiveMqReceiver extends BaseTest {
        @Autowired
        private SimpleJmsTemplate receiverJmsTemplate;
        @Autowired
        private ActiveMQQueue activeMQQueue;
    
        /** * 如果session事物设置为true,receiver直接将sessioin进行commit, * <p> * 如果设置为false,receiver方法会直接判断进行消息确认,无法做到手动的消息确认,所以一旦发生异常,这条消息不会回到消息队列中 * <p> * session的提交可以认为是消息确认收到 * * @throws JMSException */ @Test
        public void recerver() throws JMSException {
            int i = 1;
            int j = 0;
            String eq = null;
            while (true) {
                i++;
                TextMessage message = (TextMessage) receiverJmsTemplate.receive(activeMQQueue);
                if (null != message) {
                    String messageText = message.getText();
                    eq = messageText;
    
                    System.out.println("收到消息==================" + messageText);
                    if (messageText.equals("我是第2个")) {
                        try {
                            throw new RuntimeException("Exception");
                        } catch (Exception e) {
                            System.out.println("第" + j + "次接收");
                            j++;
                            receiverJmsTemplate.getSessionToUse().recover();
                        }
                    }
                    receiverJmsTemplate.msgAckAndcloseSession(message);
                } else {
                    System.out.print("超时10秒");
                    break;
                }
            }
        }
    }

# 6:测试结果 #

收到消息==================我是第1个  
收到消息==================我是第2个  
第0次接收  
收到消息==================我是第2个  
第1次接收  
收到消息==================我是第2个  
第2次接收  
收到消息==================我是第2个  
第3次接收  
收到消息==================我是第2个  
第4次接收  
收到消息==================我是第2个  
第5次接收  
收到消息==================我是第3个  
收到消息==================我是第4个  
收到消息==================我是第5个  
收到消息==================我是第6个  
收到消息==================我是第7个  
收到消息==================我是第8个  
收到消息==================我是第9个  
收到消息==================我是第10个

### 注:这里第一次是默认的,后五次是配置的重新发送,当五次重新发送之后还是失败,那么将会进入死信队列,看下一步 ###

# 7:死信队列结果 #

![SouthEast][]

###     ###

#     #

[SouthEast]: /images/20220717/ba991567e59c484ea22630fdfa55c969.png