版本 apache-activemq-5.15.3
1、消息過期設置
參數詳情
1）message過期則客戶端不能接收
2）ttlCeiling：表示過期時間上限（程序寫的過期時間不能超過此時間，超過則以此時間為准）
3）zeroExpirationOverride：表示過期時間（給未分配過期時間的消息分配過期時間）
配置示例
-
<broker>
-
...
-
<plugins>
-
<!-- 86,400,000ms = 1 day -->
-
<timeStampingBrokerPluginttlCeiling="30000" zeroExpirationOverride="30000" />
-
</plugins>
-
...
-
</broker>
2、死信隊列設置
消息過期后會進入死信隊列，如不想拋棄死信隊列，默認進入ACTIVEMQ.DLQ隊列，且不會自動清除；對於過期的消息進入死信隊列還有一些可選的策略:放入各自的死信通道、保存在一個共享的隊列（默認），且可以設置是否將過期消息放入隊列的開關以及死信隊列消息過期時間。
1）直接拋棄死信隊列
AcitveMQ提供了一個便捷的插件：DiscardingDLQBrokerPlugin，來拋棄DeadLetter。如果開發者不需要關心DeadLetter，可以使用此策略。
配置示例
-
<broker>
-
...
-
<plugins>
-
<!-- 丟棄所有死信-->
-
<discardingDLQBrokerPlugindropAll="true" dropTemporaryTopics="true" dropTemporaryQueues="true" />
-
<!-- 丟棄指定死信-->
-
<!-- <discardingDLQBrokerPlugindropOnly="MY.EXAMPLE.TOPIC.29 MY.EXAMPLE.QUEUE.87" reportInterval="1000" />-->
-
<!--使用丟棄正則匹配到死信-->
-
<!--<discardingDLQBrokerPlugindropOnly="MY.EXAMPLE.TOPIC.[0-9]{3} MY.EXAMPLE.QUEUE.[0-9]{3}" reportInterval="3000"/>-->
-
</plugins>
-
...
-
</broker>
2）定時拋棄死信隊列
默認情況下，ActiveMQ永遠不會過期發送到DLQ的消息。但是，從ActiveMQ 5.12開始，deadLetterStrategy支持expiration屬性，其值以毫秒為單位。
配置示例
-
<policyEntryqueue=">"…>
-
...
-
<deadLetterStrategy>
-
<sharedDeadLetterStrategy processExpired="true" expiration="30000"/>
-
</deadLetterStrategy>
-
...
-
</policyEntry>
3、慢消費者策略設置
Broker將會啟動一個后台線程用來檢測所有的慢速消費者，並定期關閉關閉它們;中斷慢速消費者，慢速消費將會被關閉。abortConnection是否關閉連接;如果慢速消費者最后一個ACK距離現在的時間間隔超過閥maxTimeSinceLastAck，則中斷慢速消費者。
配置示例
-
<policyEntryqueue=">"…>
-
…
-
<slowConsumerStrategy>
-
<abortSlowConsumerStrategyabortConnection="false"/>
<!--不關閉底層鏈接-->
-
</slowConsumerStrategy>
-
…
-
</policyEntry>
4、測試
1）配置activemq.xml；
2）批量消息發送；
3）等待若干秒，消息減少；
4）等待若干秒，消息隊列清除（設置了拋棄死信隊列）；
5）等待若干秒，消息隊列清除（未設置拋棄死信隊列，設置了死信隊列過期時間）；
6）等待若干秒，消息隊列清除（不拋棄死信隊列，死信隊列不過期）；
5、參考網址
https://my.oschina.net/coderedrain/blog/724943?utm_source=tuicool&utm_medium=referral
http://ask.csdn.net/questions/376817
http://activemq.apache.org/message-redelivery-and-dlq-handling.html
完整配置
-
<!--
-
Licensed to the Apache Software Foundation (ASF) under one or more
-
contributor license agreements. See the NOTICE file distributed with
-
this work for additional information regarding copyright ownership.
-
The ASF licenses this file to You under the Apache License, Version 2.0
-
(the "License"); you may not use this file except in compliance with
-
the License. You may obtain a copy of the License at
-
-
http://www.apache.org/licenses/LICENSE-2.0
-
-
Unless required by applicable law or agreed to in writing, software
-
distributed under the License is distributed on an "AS IS" BASIS,
-
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-
See the License for the specific language governing permissions and
-
limitations under the License.
-
-->
-
<!-- START SNIPPET: example -->
-
<beans
-
xmlns=
"http://www.springframework.org/schema/beans"
-
xmlns:xsi=
"http://www.w3.org/2001/XMLSchema-instance"
-
xsi:schemaLocation=
"http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
-
http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd">
-
-
<!-- Allows us to use system properties as variables in this configuration file -->
-
<bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
-
<property name="locations">
-
<value>file:${activemq.conf}/credentials.properties
</value>
-
</property>
-
</bean>
-
-
<!-- Allows accessing the server log -->
-
<bean id="logQuery" class="io.fabric8.insight.log.log4j.Log4jLogQuery"
-
lazy-init=
"false"
scope=
"singleton"
-
init-method=
"start"
destroy-method=
"stop">
-
</bean>
-
-
<!--
-
The <broker> element is used to configure the ActiveMQ broker.
-
-->
-
<broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost" dataDirectory="${activemq.data}" schedulePeriodForDestinationPurge="10000">
-
<destinationPolicy>
-
<policyMap>
-
<policyEntries>
-
<policyEntry queue=">" gcInactiveDestinations="true" inactiveTimoutBeforeGC="30000">
-
<deadLetterStrategy>
-
<sharedDeadLetterStrategy processExpired="true" expiration="30000"/>
-
</deadLetterStrategy>
-
</policyEntry>
-
<policyEntry topic=">" >
-
<!-- The constantPendingMessageLimitStrategy is used to prevent
-
slow topic consumers to block producers and affect other consumers
-
by limiting the number of messages that are retained
-
For more information, see:
-
http://activemq.apache.org/slow-consumer-handling.html
-
-->
-
<pendingMessageLimitStrategy>
-
<constantPendingMessageLimitStrategy limit="1000"/>
-
</pendingMessageLimitStrategy>
-
</policyEntry>
-
</policyEntries>
-
</policyMap>
-
</destinationPolicy>
-
-
<!--
-
The managementContext is used to configure how ActiveMQ is exposed in
-
JMX. By default, ActiveMQ uses the MBean server that is started by
-
the JVM. For more information, see:
-
http://activemq.apache.org/jmx.html
-
-->
-
<managementContext>
-
<managementContext createConnector="false"/>
-
</managementContext>
-
-
<!--
-
Configure message persistence for the broker. The default persistence
-
mechanism is the KahaDB store (identified by the kahaDB tag).
-
For more information, see:
-
-
http://activemq.apache.org/persistence.html
-
-->
-
<persistenceAdapter>
-
<kahaDB directory="${activemq.data}/kahadb"/>
-
</persistenceAdapter>
-
<!--
-
The systemUsage controls the maximum amount of space the broker will
-
use before disabling caching and/or slowing down producers. For more information, see:
-
http://activemq.apache.org/producer-flow-control.html
-
-->
-
<systemUsage>
-
<systemUsage>
-
<memoryUsage>
-
<memoryUsage percentOfJvmHeap="70" />
-
</memoryUsage>
-
<storeUsage>
-
<storeUsage limit="100 gb"/>
-
</storeUsage>
-
<tempUsage>
-
<tempUsage limit="50 gb"/>
-
</tempUsage>
-
</systemUsage>
-
</systemUsage>
-
-
<!--
-
The transport connectors expose ActiveMQ over a given protocol to
-
clients and other brokers. For more information, see:
-
http://activemq.apache.org/configuring-transports.html
-
-->
-
<transportConnectors>
-
<!-- DOS protection, limit concurrent connections to 1000 and frame size to 100MB -->
-
<transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
-
<transportConnector name="amqp" uri="amqp://0.0.0.0:5672?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
-
<transportConnector name="stomp" uri="stomp://0.0.0.0:61613?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
-
<transportConnector name="mqtt" uri="mqtt://0.0.0.0:1883?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
-
<transportConnector name="ws" uri="ws://0.0.0.0:61614?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
-
</transportConnectors>
-
-
<!-- destroy the spring context on shutdown to stop jetty -->
-
<shutdownHooks>
-
<bean xmlns="http://www.springframework.org/schema/beans" class="org.apache.activemq.hooks.SpringContextHook" />
-
</shutdownHooks>
-
-
<plugins>
-
<!-- 86,400,000ms = 1 day -->
-
<timeStampingBrokerPlugin ttlCeiling="30000" zeroExpirationOverride="30000" />
-
<!-- <discardingDLQBrokerPlugin dropAll="true" dropTemporaryTopics="true" dropTemporaryQueues="true" />-->
-
</plugins>
-
</broker>
-
-
<!--
-
Enable web consoles, REST and Ajax APIs and demos
-
The web consoles requires by default login, you can disable this in the jetty.xml file
-
Take a look at ${ACTIVEMQ_HOME}/conf/jetty.xml for more details
-
-->
-
<import resource="jetty.xml"/>
-
</beans>
-
<!-- END SNIPPET: example -->