spring-boot - 在 Spring Boot 中为嵌入式 ActiveMQ 更改 Activemq RedeliveryPolicy

Question

如何在使用 Spring Boot 时更改嵌入式 ActiveMQ 的重新传递策略？我尝试在 DefaultJmsListenerContainerFactory 上指定 FixedBackOff 但它没有帮助。下面是我用来初始化 jms 工厂 bean 的代码。我有一个消息使用者处理队列中的传入消息。由于资源不可用，在处理过程中，我抛出了一个已检查的异常。我希望在固定时间间隔后重新传送消息以进行处理。

Spring Boot:1.5.7.发布

Java:1.7

@Bean
public JmsListenerContainerFactory<?> publishFactory(ConnectionFactory connectionFactory,
                                                DefaultJmsListenerContainerFactoryConfigurer configurer) {
    DefaultJmsListenerContainerFactory factory = 
        new DefaultJmsListenerContainerFactory();

    factory.setBackOff(new FixedBackOff(5000, 5));

    // This provides all boot's default to this factory, including the message converter
    configurer.configure(factory, connectionFactory);

    // You could still override some of Boot's default if necessary.
    factory.setErrorHandler(new ErrorHandler() {

        @Override
        public void handleError(Throwable t) {
            LOG.error("Error occured in JMS transaction.", t);
        }

    }); 
    return factory;
}

消费者代码:

@JmsListener(destination = "PublishQueue", containerFactory = "publishFactory")
@Transactional
public void receiveMessage(PublishData publishData) {
    LOG.info("Processing incoming message on publish queue with transaction id: " + publishData.getTransactionId());

    PublishUser user = new PublishUser();
    user.setPriority(1);
    user.setUserId(publishData.getUserId());

    LOG.trace("Trying to enroll in the publish lock queue for user: " + user);
    PublishLockQueue lockQueue = publishLockQueueService.createLock(user);
    if (lockQueue == null)
        throw new RuntimeException("Unable to create lock for publish");
    LOG.trace("Publish lock queue obtained with lock queue id: " + lockQueue.getId());

    try {
        publishLockQueueService.acquireLock(lockQueue.getId());
        LOG.trace("Acquired publish lock.");
    }
    catch (PublishLockQueueServiceException pex) {
        throw new RuntimeException(pex);
    }

    try {
        publishService.publish(publishData, lockQueue.getId());
        LOG.trace("Completed publish of changes.");

        sendPublishSuccessNotification(publishData);
    }
    finally {
        LOG.trace("Trying to release lock to publish.");
        publishLockQueueService.releaseLock(lockQueue.getId());
    }

    LOG.info("Publish has been completed for transaction id: " + publishData.getTransactionId());
}

Answer 1

@claus 回答:我测试过它可以工作:

Its the consumer, you need to use transacted acknowledge mode to let the consumer rollback on exception and let ActiveMQ be able to re-deliver the message to the same consumer or another consumer if you have multiple consumers running. You can however configure redelivery options on the ActiveMQ such as backoff etc. The error handler above is just a noop listener which cannot do very much other than logging