快捷導(dǎo)航

Springboot實(shí)現(xiàn)Activemq死信隊(duì)列詳解

更新時(shí)間：2023年12月13日 08:52:32 作者：demon7552003

這篇文章主要介紹了Springboot實(shí)現(xiàn)Activemq死信隊(duì)列詳解,Activemq服務(wù)端配置重新投遞次數(shù)超過?MaximumRedeliveries?,則會(huì)進(jìn)入死信隊(duì)列,默認(rèn)情況,有一個(gè)死信隊(duì)列：AcitveMQ.DLQ,所有的消息都投遞到此隊(duì)列,包括過期消息,重投遞失敗消息,需要的朋友可以參考下

死信隊(duì)列是什么

當(dāng)消息不能重投遞或者消息過期，會(huì)被移到死信隊(duì)列中，由管理員消費(fèi)。

可以進(jìn)行以下操作：

delete：刪除記錄
retry：重新投遞
copy：復(fù)制到一個(gè)選擇的隊(duì)列中。
move：移動(dòng)到一個(gè)選擇的隊(duì)列中。

什么情況下消息會(huì)重投遞

消息重投遞的常用場(chǎng)景：

事務(wù)回滾
事務(wù)提交前close
A client is using CLIENT_ACKNOWLEDGE on a Session and calls recover() on that Session.

重新投遞策略

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

死信隊(duì)列

重新投遞次數(shù)超過 MaximumRedeliveries ，則會(huì)進(jìn)入死信隊(duì)列。

默認(rèn)情況，有一個(gè)死信隊(duì)列：AcitveMQ.DLQ，所有的消息都投遞到此隊(duì)列，包括過期消息，重投遞失敗消息。

Spring整合

Activemq服務(wù)端配置

重新投遞次數(shù)超過 MaximumRedeliveries ，則會(huì)進(jìn)入死信隊(duì)列。

默認(rèn)情況，有一個(gè)死信隊(duì)列：AcitveMQ.DLQ，所有的消息都投遞到此隊(duì)列，包括過期消息，重投遞失敗消息。

配置個(gè)性化死信隊(duì)列。

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

可以單獨(dú)設(shè)置重投遞策略

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

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

                //設(shè)置了隊(duì)列的 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");
            }
        };
    }

觸發(fā)重投遞

代碼回滾

 @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){

        }
    }

拋出異常自動(dòng)回滾

由于默認(rèn)是開啟事務(wù)的，因此拋出異常，會(huì)自動(dòng)觸發(fā)回滾。

@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類異常會(huì)回滾。
		catch (JMSException | RuntimeException | Error ex) {
      //自動(dòng)回滾
			rollbackOnExceptionIfNecessary(session, ex);
			throw ex;
		}
     //自動(dòng)提交
		commitIfNecessary(session, message);
	}

其他的Exception 都會(huì)被包裝成ListenerExecutionFailedException，它是JMSException的子類，所以所有異常都會(huì)導(dǎo)致回滾。

到此這篇關(guān)于Springboot實(shí)現(xiàn)Activemq死信隊(duì)列詳解的文章就介紹到這了,更多相關(guān)Activemq死信隊(duì)列內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: