Springboot实现Activemq死信队列详解

RedeliveryPolicy policy = connection.getRedeliveryPolicy();
policy.setInitialRedeliveryDelay(500);
policy.setBackOffMultiplier(2);
policy.setUseExponentialBackOff(true);
policy.setMaximumRedeliveries(2);

<destinationPolicy>
    <policyMap>
        <policyEntries>
            <policyEntry queue=">">
                <deadLetterStrategy>
                  <individualDeadLetterStrategy queuePrefix="DLQ."
                  useQueueForQueueMessages="true"
                  processExpired="false"
                  processNonPersistent="false"/>
                </deadLetterStrategy>
            </policyEntry>
        </policyEntries>
    </policyMap>
</destinationPolicy>

    @Bean
    public ActiveMQConnectionFactoryCustomizer myCustomizer2() {
        return new ActiveMQConnectionFactoryCustomizer() {

            /**
             * Customize the {@link ActiveMQConnectionFactory}.
             *
             * @param factory the factory to customize
             */
            @Override
            public void customize(ActiveMQConnectionFactory factory) {

                //设置了队列的 policy。
                ActiveMQQueue q = new ActiveMQQueue("queue.dlq.test");
                RedeliveryPolicy policy = new RedeliveryPolicy();
                policy.setMaximumRedeliveries(3);
                factory.getRedeliveryPolicyMap().put(q,policy);

                System.out.println("Customizer 2");
            }
        };
    }

 @JmsListener(destination = "queue.dlq.test" ,id = "test")
    public void consume(String param, Session session) {

        try {
            System.out.println(session.getAcknowledgeMode());
            System.out.println(param);
            //回滚，则重投递
            session.rollback();
        }catch (Exception ex){

        }
    }

@JmsListener(destination = "queue.dlq.test2",id="test2")
    public void consume2(String param, Session session) {
        System.out.println(param);
        throw new RuntimeException("xxxx");
    }

//AbstractMessageListenerContainer
protected void doExecuteListener(Session session, Message message) throws JMSException {
		if (!isAcceptMessagesWhileStopping() && !isRunning()) {
			if (logger.isWarnEnabled()) {
				logger.warn("Rejecting received message because of the listener container " +
						"having been stopped in the meantime: " + message);
			}
			rollbackIfNecessary(session);
			throw new MessageRejectedWhileStoppingException();
		}

		try {
			invokeListener(session, message);
		}
     //JMSException，RuntimeException，Error  这3类异常会回滚。
		catch (JMSException | RuntimeException | Error ex) {
      //自动回滚
			rollbackOnExceptionIfNecessary(session, ex);
			throw ex;
		}
     //自动提交
		commitIfNecessary(session, message);
	}