ActiveMQ Artemis生产环境队列消息指数级重复生成问题求助
问题:ActiveMQ Artemis 消息指数级重复生成
生产环境使用ActiveMQ Artemis 2.18.0搭配.NET Core 3.1,组件为Apache.NMS.ActiveMQ.NetStd 1.8.0和Apache.NMS.AMQP 1.8.1。目前存在一个每1-2个月触发一次的严重问题:队列中的消息会指数级重复生成。
我们的处理逻辑是:从队列接收通知后,若请求暂时无法处理(如支付服务商要求稍后重试),会手动设置_AMQ_SCHED_DELIVERY将消息重新入队,而非依赖Broker自动处理重发延迟——因为仅希望特定场景下才触发重发。但此时会并行收到2-3条相同通知(有接收日志记录),最终队列被大量重复消息填满,不得不关闭ActiveMQ Artemis并手动清空队列。
消息消费采用事务机制,收到消息后立即调用message.acknowledge(),随后执行session.commit()。队列均为动态创建,未在broker.xml中配置。
broker.xml 配置
<configuration xmlns="urn:activemq" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:xi="http://www.w3.org/2001/XInclude" xsi:schemaLocation="urn:activemq /schema/artemis-configuration.xsd"> <core xmlns="urn:activemq:core" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="urn:activemq:core "> <name>ac3bc22d67ca</name> <persistence-enabled>true</persistence-enabled> <journal-type>ASYNCIO</journal-type> <paging-directory>data/paging</paging-directory> <bindings-directory>data/bindings</bindings-directory> <journal-directory>data/journal</journal-directory> <large-messages-directory>data/large-messages</large-messages-directory> <journal-datasync>true</journal-datasync> <journal-min-files>2</journal-min-files> <journal-pool-files>10</journal-pool-files> <journal-device-block-size>4096</journal-device-block-size> <journal-file-size>10M</journal-file-size> <journal-buffer-timeout>28000</journal-buffer-timeout> <journal-max-io>4096</journal-max-io> <disk-scan-period>5000</disk-scan-period> <max-disk-usage>99</max-disk-usage> <critical-analyzer>true</critical-analyzer> <critical-analyzer-timeout>120000</critical-analyzer-timeout> <critical-analyzer-check-period>60000</critical-analyzer-check-period> <critical-analyzer-policy>HALT</critical-analyzer-policy> <page-sync-timeout>152000</page-sync-timeout> <global-max-messages>-1</global-max-messages> <acceptors> <acceptor name="artemis">tcp://0.0.0.0:61616?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;amqpMinLargeMessageSize=102400;protocols=CORE,AMQP,STOMP,HORNETQ,MQTT,OPENWIRE;useEpoll=true;amqpCredits=1000;amqpLowCredits=300;amqpDuplicateDetection=true;supportAdvisory=false;suppressInternalManagementObjects=false</acceptor> <acceptor name="amqp">tcp://0.0.0.0:5672?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=AMQP;useEpoll=true;amqpCredits=1000;amqpLowCredits=300;amqpMinLargeMessageSize=102400;amqpDuplicateDetection=true</acceptor> <acceptor name="stomp">tcp://0.0.0.0:61613?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=STOMP;useEpoll=true</acceptor> <acceptor name="hornetq">tcp://0.0.0.0:5445?anycastPrefix=jms.queue.;multicastPrefix=jms.topic.;protocols=HORNETQ,STOMP;useEpoll=true</acceptor> <acceptor name="mqtt">tcp://0.0.0.0:1883?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=MQTT;useEpoll=true</acceptor> </acceptors> <security-settings> <security-setting match="#"> <permission type="createNonDurableQueue" roles="amq"/> <permission type="deleteNonDurableQueue" roles="amq"/> <permission type="createDurableQueue" roles="amq"/> <permission type="deleteDurableQueue" roles="amq"/> <permission type="createAddress" roles="amq"/> <permission type="deleteAddress" roles="amq"/> <permission type="consume" roles="amq"/> <permission type="browse" roles="amq"/> <permission type="send" roles="amq"/> <permission type="manage" roles="amq"/> </security-setting> </security-settings> <address-settings> <address-setting match="activemq.management#"> <dead-letter-address>DLQ</dead-letter-address> <expiry-address>ExpiryQueue</expiry-address> <redelivery-delay>0</redelivery-delay> <max-size-bytes>-1</max-size-bytes> <message-counter-history-day-limit>10</message-counter-history-day-limit> <address-full-policy>PAGE</address-full-policy> <auto-create-queues>true</auto-create-queues> <auto-create-addresses>true</auto-create-addresses> </address-setting> <address-setting match="#"> <dead-letter-address>DLQ</dead-letter-address> <expiry-address>ExpiryQueue</expiry-address> <redelivery-delay>0</redelivery-delay> <max-size-bytes>-1</max-size-bytes> <max-size-messages>-1</max-size-messages> <message-counter-history-day-limit>10</message-counter-history-day-limit> <address-full-policy>PAGE</address-full-policy> <auto-create-queues>true</auto-create-queues> <auto-create-addresses>true</auto-create-addresses> <auto-delete-queues>false</auto-delete-queues> <auto-delete-addresses>false</auto-delete-addresses> </address-setting> </address-settings> <addresses> <address name="DLQ"> <anycast> <queue name="DLQ"/> </anycast> </address> <address name="ExpiryQueue"> <anycast> <queue name="ExpiryQueue"/> </anycast> </address> </addresses> </core> </configuration>
简化后的消费者代码
consumer.Listener += this.OnReceive; private void OnReceive(IMessage message) { var textMessage = message as ITextMessage; message.Acknowledge(); this.Session.Commit(); // here comes the real logic for handling the request. }
请问是否有遇到过类似问题的同行,能否提供解决方案?
内容的提问来源于stack exchange,提问作者Tettsuan
相关产品推荐
相关产品推荐

