1

我正在尝试使用 ActiveMQ 实现死信队列。不幸的是，这方面的文档在某些方面相当模糊，我似乎无法正确设置所有内容。

我配置了以下 Bean：

@Bean
public JmsTemplate createJMSTemplate() {
    logger.info("createJMSTemplate");
    JmsTemplate jmsTemplate = new JmsTemplate(getActiveMQConnectionFactory());
    jmsTemplate.setDefaultDestinationName(queue);
    jmsTemplate.setDeliveryPersistent(true);
    jmsTemplate.setDeliveryMode(DeliveryMode.PERSISTENT);
    return jmsTemplate;
}

@Bean
public DefaultJmsListenerContainerFactory jmsListenerContainerFactory() {
    DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
    factory.setConnectionFactory(getActiveMQConnectionFactory());
    factory.setConcurrency("1-10");
    factory.setSessionTransacted(false);
    factory.setSessionAcknowledgeMode(Session.CLIENT_ACKNOWLEDGE);
    return factory;
}

@Bean
public ConnectionFactory getActiveMQConnectionFactory() {
    // Configure the ActiveMQConnectionFactory
    ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory();
    activeMQConnectionFactory.setBrokerURL("tcp://127.0.0.1:61616");
    activeMQConnectionFactory.setTrustedPackages(Arrays.asList("com.company"));

    // Configure the redeliver policy and the dead letter queue
    RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
    redeliveryPolicy.setInitialRedeliveryDelay(0);
    redeliveryPolicy.setRedeliveryDelay(10000);
    redeliveryPolicy.setUseExponentialBackOff(true);
    redeliveryPolicy.setMaximumRedeliveries(3);
    RedeliveryPolicyMap redeliveryPolicyMap = activeMQConnectionFactory.getRedeliveryPolicyMap();
    redeliveryPolicyMap.put(new ActiveMQQueue(queue), redeliveryPolicy);
    activeMQConnectionFactory.setRedeliveryPolicy(redeliveryPolicy);

    return activeMQConnectionFactory;
}

这是我的接收代码：

@Autowired
private ConnectionFactory connectionFactory;

private static Logger logger = LoggerFactory.getLogger(QueueReceiver.class);
private Connection connection;
private Session session;
private SegmentReceiver callback;

@PostConstruct
private void init() throws JMSException, InterruptedException {
    logger.info("Initializing QueueReceiver...");
    this.connection = connectionFactory.createConnection();
    this.session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
    Queue q = session.createQueue(queue);
    logger.info("Creating consumer for queue '{}'", q.getQueueName());
    MessageConsumer consumer = session.createConsumer(q);
    this.callback = new SegmentReceiver();
    consumer.setMessageListener(callback);
    this.connection.start();
}

@PreDestroy
private void destroy() throws JMSException {
    logger.info("Destroying QueueReceiver...");
    this.session.close();
    this.connection.close();
}

private class SegmentReceiver implements MessageListener {

    @Override
    public void onMessage(Message message) {
        logger.info("onMessage");
        try {
            TextMessage textMessage = (TextMessage) message;
            Segment segment = Segment.fromJSON(textMessage.getText());
            if (segment.shouldFail()) {
                throw new IOException("This segment is expected to fail");
            }
            System.out.println(segment.getText());
            message.acknowledge();
        }
        catch(IOException | JMSException exception) {
            logger.error(exception.toString());
            try {
                QueueReceiver.this.session.rollback();
            } catch (JMSException e) {
                logger.error(e.toString());
            }
            throw new RuntimeException(exception);
        }
    }

}

然而，什么也没有发生。我正在使用默认配置的开箱即用 Apache ActiveMQ 5.14.2 设置。我在这里想念什么？

4

2 回答 2

2

因为您使用 的this.session = connection.createSession(false,Session.CLIENT_ACKNOWLEDGE);
调用message.acknowledge();与调用相同session.acknowledge();。

要使 ActiveMQ 重新交付成功地与您的配置一起工作，有一些可能只需进行最小的更改：

  1. 打电话QueueReceiver.this.session.recover();
    代替打电话 QueueReceiver.this.session.rollback();

无效 org.apache.activemq.ActiveMQSession.recover() 抛出 JMSException

在此会话中停止消息传递，并使用最旧的未确认消息重新开始消息传递。

所有消费者都按顺序传递消息。确认收到的消息会自动确认已传递给客户端的所有消息。

重新启动会话会导致它采取以下操作： •停止消息传递 •将所有可能已传递但未确认的消息标记为“重新传递” •重新启动传递序列，包括先前已传递的所有未确认消息。重新传递的消息不必完全按照其原始传递顺序传递。

  1. 使用 this.session = connection.createSession(false, org.apache.activemq.ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE); 和调用 ((org.apache.activemq.command.ActiveMQMessage) message ).acknowledge();，注意不调用这个方法就像回滚，意味着消息没有被确认并且在方法中抛出异常将onMessage调用org.apache.activemq.ActiveMQMessageConsumer.rollback()。QueueReceiver.this.consumer.rollback();

  2. 只需调用QueueReceiver.this.consumer.rollback(); org.apache.activemq.ActiveMQMessageConsumer.rollback()代替调用QueueReceiver.this.session.rollback();

于 2016-12-16T13:42:52.237 回答
0

所以事实证明这是一个问题的组合：

  • 会话确认模式需要设置为ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE
  • 我正在使用session.recover()而不是rollback()

  • ActiveMQ 代理未正确配置。我需要将此位添加到 activemq.xml 配置文件中（将其放在<broker>标记下）。

        <destinationPolicy>
    <policyMap>
      <policyEntries>
        <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>
        <!-- Set the following policy on all queues using the '>' wildcard -->
        <policyEntry queue=">">
            <deadLetterStrategy>
                <!--
                  Use the prefix 'DLQ.' for the destination name, and make
                  the DLQ a queue rather than a topic
                -->
                <individualDeadLetterStrategy queuePrefix="DLQ." useQueueForQueueMessages="true"/>
            </deadLetterStrategy>
        </policyEntry>
      </policyEntries>
    </policyMap>
</destinationPolicy>

  • 确保您没有激活任何redeliveryPlugin可能与您的 ActiveMQConnectionFactory 配置混淆的东西。

于 2016-12-19T12:59:37.807 回答