java - 从虚拟主题接收相同消息的多个队列仅为一个队列创建死信条目

Question

我正在使用虚拟目标在 ActiveMQ 5.15.13 中实现发布订阅模型。

我有一个虚拟主题VirtualTopic并且有两个队列绑定(bind)到它。每个队列都有自己的重新传递策略。假设在处理消息时出现异常，Queue 1 将重试消息 2 次，而 Queue 2 将重试消息 3 次。重试后消息将被发送到死信队列。我还使用单独死信队列策略，以便每个队列都有自己的死信队列。

我观察到，当一条消息发送到VirtualTopic时，具有相同消息id的消息会被传递到两个队列。我面临一个问题，如果两个队列的消费者都无法成功处理消息。发往Queue 1的消息在重试2次后被移至死信队列。但是Queue 2没有死信队列，尽管Queue 2中的消息重试了3次。

这是预期的行为吗？

代码:

public class ActiveMQRedelivery {

    private final ActiveMQConnectionFactory factory;

    public ActiveMQRedelivery(String brokerUrl) {
        factory = new ActiveMQConnectionFactory(brokerUrl);
        factory.setUserName("admin");
        factory.setPassword("password");
        factory.setAlwaysSyncSend(false);
    }

    public void publish(String topicAddress, String message) {
        final String topicName = "VirtualTopic." + topicAddress;
        try {
            final Connection producerConnection = factory.createConnection();
            producerConnection.start();
            final Session producerSession = producerConnection.createSession(false, AUTO_ACKNOWLEDGE);
            final MessageProducer producer = producerSession.createProducer(null);
            final TextMessage textMessage = producerSession.createTextMessage(message);
            final Topic topic = producerSession.createTopic(topicName);
            producer.send(topic, textMessage, PERSISTENT, DEFAULT_PRIORITY, DEFAULT_TIME_TO_LIVE);
        } catch (JMSException e) {
            throw new RuntimeException("Message could not be published", e);
        }

    }

    public void initializeConsumer(String queueName, String topicAddress, int numOfRetry) throws JMSException {
        factory.getRedeliveryPolicyMap().put(new ActiveMQQueue("*." + queueName + ".>"),
                getRedeliveryPolicy(numOfRetry));
        Connection connection = factory.createConnection();
        connection.start();
        final Session consumerSession = connection.createSession(false, CLIENT_ACKNOWLEDGE);
        final Queue queue = consumerSession.createQueue("Consumer." + queueName +
                ".VirtualTopic." + topicAddress);
        final MessageConsumer consumer = consumerSession.createConsumer(queue);
        consumer.setMessageListener(message -> {
            try {
                    System.out.println("in listener --- " + ((ActiveMQDestination)message.getJMSDestination()).getPhysicalName());
                    consumerSession.recover();
            } catch (JMSException e) {
                e.printStackTrace();
            }
        });
    }

    private RedeliveryPolicy getRedeliveryPolicy(int numOfRetry) {
        final RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
        redeliveryPolicy.setInitialRedeliveryDelay(0);
        redeliveryPolicy.setMaximumRedeliveries(numOfRetry);
        redeliveryPolicy.setMaximumRedeliveryDelay(-1);
        redeliveryPolicy.setRedeliveryDelay(0);
        return redeliveryPolicy;
    }

}

测试:

public class ActiveMQRedeliveryTest {

    private static final String brokerUrl = "tcp://0.0.0.0:61616";
    private ActiveMQRedelivery activeMQRedelivery;

    @Before
    public void setUp() throws Exception {
        activeMQRedelivery = new ActiveMQRedelivery(brokerUrl);
    }

    @Test
    public void testMessageRedeliveries() throws Exception {
        String topicAddress = "testTopic";
        activeMQRedelivery.initializeConsumer("queue1", topicAddress, 2);
        activeMQRedelivery.initializeConsumer("queue2", topicAddress, 3);
        activeMQRedelivery.publish(topicAddress, "TestMessage");
        Thread.sleep(3000);
    }

    @After
    public void tearDown() throws Exception {
    }
}

Answer 1

我最近遇到了这个问题。要解决此问题，需要将 2 个属性添加到 individualDeadLetterStrategy 中，如下所示

<deadLetterStrategy>
        <individualDeadLetterStrategy destinationPerDurableSubscriber="true" enableAudit="false" queuePrefix="DLQ." useQueueForQueueMessages="true"/>
</deadLetterStrategy>

属性说明:

destinationPerDurableSubscriber - 为每个持久订阅者启用单独的目标。

enableAudit - 死信策略默认启用消息审核。这可以防止将重复的消息添加到配置的 DLQ 中。启用该属性后，当destinationPerDurableSubscriber 属性设置为 true 时，未向同一主题的多个订阅者传递的同一消息将仅放置在其中一个订阅者 DLQ 上，即两个消费者未能确认该主题的同一消息。主题，该消息将仅放置在一个消费者的 DLQ 上，而不是另一个消费者的 DLQ 上。