Я использую:
По сути, у меня есть приложение SpringBoot с маршрутом Apache Camel, которое принимает сообщения из ActiveMQ с транзакциями. Мне нужно установить RedeliveryPolicy в ActiveMQ, чтобы при возникновении ошибки при обработке сообщение повторялось несколько раз.
Я создал класс конфигурации с bean-компонентами для ActiveMQ, транзакции работают, как задумано, но RedeliveryPolicy не работает. Кто-нибудь может помочь мне понять, что с этим не так?
Вот вывод журнала для сообщения об ошибке:
2018-10-23 10:35:28.005 DEBUG 10524 --- [mer[entryQueue]] o.a.c.s.spi.TransactionErrorHandler : Transaction begin (0x35d60381) redelivered(false) for (MessageId: ID:EPIC-LAP-25-50304-1540306817804-4:3:1:1:2 on ExchangeId: ID-EPIC-LAP-25-1540312510586-0-1)) 2018-10-23 10:35:28.020 DEBUG 10524 --- [mer[entryQueue]] o.apache.camel.processor.SendProcessor : >>>> direct://middle Exchange[ID-EPIC-LAP-25-1540312510586-0-1] 2018-10-23 10:35:28.375 DEBUG 10524 --- [mer[entryQueue]] o.a.camel.processor.DefaultErrorHandler : Failed delivery for (MessageId: ID:EPIC-LAP-25-50304-1540306817804-4:3:1:1:2 on ExchangeId: ID-EPIC-LAP-25-1540312510586-0-1). On delivery attempt: 0 caught: java.lang.RuntimeException: ExceptionTest: Order Failed 2018-10-23 10:35:28.390 ERROR 10524 --- [mer[entryQueue]] o.a.camel.processor.DefaultErrorHandler : Failed delivery for (MessageId: ID:EPIC-LAP-25-50304-1540306817804-4:3:1:1:2 on ExchangeId: ID-EPIC-LAP-25-1540312510586-0-1). Exhausted after delivery attempt: 1 caught: java.lang.RuntimeException: ExceptionTest: Order Failed
Вот мой класс конфигурации для ActiveMQ:
import org.apache.activemq.ActiveMQConnectionFactory
import org.apache.activemq.RedeliveryPolicy
import org.apache.activemq.camel.component.ActiveMQComponent
import org.springframework.beans.factory.annotation.Qualifier
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.jms.connection.JmsTransactionManager
import javax.jms.DeliveryMode
@Configuration
class ActiveMQConfiguration {
@Bean
ActiveMQConnectionFactory activeMQConnectionFactory() {
ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory()
activeMQConnectionFactory.brokerURL = 'tcp://localhost:61616'
activeMQConnectionFactory.userName = 'admin'
activeMQConnectionFactory.password = 'admin'
RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy()
redeliveryPolicy.maximumRedeliveries = 3
redeliveryPolicy.redeliveryDelay = 150L
redeliveryPolicy.useExponentialBackOff = true
redeliveryPolicy.backOffMultiplier = 1.5
activeMQConnectionFactory.setRedeliveryPolicy(redeliveryPolicy)
activeMQConnectionFactory
}
@Bean
ActiveMQComponent activeMQComponent(@Qualifier('activeMQConnectionFactory')ActiveMQConnectionFactory activeMQConnectionFactory) {
ActiveMQComponent activeMQComponent = new ActiveMQComponent()
activeMQComponent.connectionFactory = activeMQConnectionFactory
activeMQComponent.transacted = true
activeMQComponent.transactionManager = txManager()
activeMQComponent.cacheLevelName = 'CACHE_CONSUMER'
activeMQComponent.lazyCreateTransactionManager = false
activeMQComponent.deliveryMode = DeliveryMode.PERSISTENT
activeMQComponent
}
@Bean
JmsTransactionManager txManager(@Qualifier('activeMQConnectionFactory') ActiveMQConnectionFactory activeMQConnectionFactory) {
JmsTransactionManager txManager = new JmsTransactionManager()
txManager.connectionFactory = activeMQConnectionFactory
txManager.rollbackOnCommitFailure = true
txManager
}
}
Не так давно у меня возникли проблемы с очередями dlq - не все параметры, заданные в коде, работали. Пришлось добавить настройки в конфиги acitvemq. Да, разделять конфиги не лучшее решение, но другого я не нашел. Ниже мой класс конфигурации для jms и пример конфигурации очереди через activemq.xml:
@Configuration
@EnableJms
public class JmsConfig {
private Environment env;
@Autowired
public void setEnv(Environment env) {
this.env = env;
}
@Bean(name = "activemq")
public ActiveMQComponent activemq(@Qualifier("activemqTransactionManager") JmsTransactionManager jmsTransactionManager,
@Qualifier("activemqConnectionFactory") ConnectionFactory connectionFactory) {
ActiveMQComponent activeMQComponent = new ActiveMQComponent();
activeMQComponent.setTransactionManager(jmsTransactionManager);
activeMQComponent.setConnectionFactory(connectionFactory);
return activeMQComponent;
}
@Bean(name = "activemqJmsTemplate")
public JmsTemplate jmsTemplate(@Qualifier("activemqConnectionFactory") ConnectionFactory connectionFactory) {
JmsTemplate template = new JmsTemplate();
template.setConnectionFactory(connectionFactory);
return template;
}
@Bean(name = "activemqTransactionPolicy")
public SpringTransactionPolicy activemqTransactionPolicy(
@Qualifier("activemqTransactionManager") JmsTransactionManager jmsTransactionManager) {
SpringTransactionPolicy springTransactionPolicy = new SpringTransactionPolicy(jmsTransactionManager);
springTransactionPolicy.setPropagationBehaviorName("PROPAGATION_REQUIRED");
return springTransactionPolicy;
}
@Bean(name = "activemqTransactionManager")
public JmsTransactionManager activemqTransactionManager(
@Qualifier("activemqConnectionFactory") ConnectionFactory connectionFactory) {
return new JmsTransactionManager(connectionFactory);
}
@Bean(name = "activemqConnectionFactory")
public ConnectionFactory connectionFactory(@Qualifier("activemqRedeliveryPolicy") RedeliveryPolicy redeliveryPolicy) {
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
connectionFactory.setBrokerURL("tcp://" + env.getProperty("queue.url"));
connectionFactory.setTrustAllPackages(true);
RedeliveryPolicyMap map = connectionFactory.getRedeliveryPolicyMap();
map.put(new ActiveMQQueue("queueName.DLQ"), redeliveryPolicy);
return connectionFactory;
}
@Bean(name = "activemqRedeliveryPolicy")
public RedeliveryPolicy redeliveryPolicy() {
RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
redeliveryPolicy.setMaximumRedeliveries(0);
return redeliveryPolicy;
}
}
Изменения в activevq.xml:
<destinationPolicy>
<policyMap>
<policyEntries>
<!--set dead letter queue for our queue. It name will be "myQueueName.DLQ"-->
<policyEntry queue = "myQueueName">
<deadLetterStrategy>
<individualDeadLetterStrategy queuePrefix = "" queueSuffix = ".DLQ"/>
</deadLetterStrategy>
</policyEntry>
<policyEntry topic = ">">
<pendingMessageLimitStrategy>
<constantPendingMessageLimitStrategy limit = "1000"/>
</pendingMessageLimitStrategy>
</policyEntry>
</policyEntries>
</policyMap>
</destinationPolicy>
<plugins>
<redeliveryPlugin fallbackToDeadLetter = "true" sendToDlqIfMaxRetriesExceeded = "true">
<redeliveryPolicyMap>
<redeliveryPolicyMap>
<redeliveryPolicyEntries>
<!--Set the redelivery delay to one hour-->
<redeliveryPolicy queue = "myQueueName.DLQ" maximumRedeliveries = "-1" redeliveryDelay = "3600000"/>
</redeliveryPolicyEntries>
</redeliveryPolicyMap>
</redeliveryPolicyMap>
</redeliveryPlugin>
</plugins>
Здесь есть две проблемы
1. У вас есть два менеджера транзакций.
Из-за следующих двух строк в вашей конфигурации компонента Camel ActiveMQ вы настраиваете два менеджера транзакций. Это источник проблем.
activeMQComponent.transacted = true // activates local JMS transactions
activeMQComponent.transactionManager = txManager() // additional tx manager
если вы просто хотите использовать транзакции из ActiveMQ, вам не нужно настраивать диспетчер транзакций Spring.
Этих двух строк вашей конфигурации достаточно для получения локальных транзакций с вашим брокером ActiveMQ.
activeMQComponent.transacted = true
activeMQComponent.lazyCreateTransactionManager = false
Значит вам стоит удалите эту строку, а также весь компонент txManager
activeMQComponent.transactionManager = txManager()
Если вы в настоящее время устанавливаете флаг транзакции в своих маршрутах Camel, вы также должны удалить его. И, как я уже писал, ваши маршруты, получаемые из ActiveMQ, по-прежнему обрабатываются, даже если вы удалите все это.
2. Повторная доставка не работает.
Вы не опубликовали свои маршруты Camel, но, судя по выводам ошибок, я предполагаю, что брокер не выполняет повторную доставку, потому что ошибка обрабатывается Camel.
Это обработчик ошибок Camel o.a.camel.processor.DefaultErrorHandler
, который срабатывает при возникновении ошибки, и поскольку он обрабатывает ошибку, сообщение передается брокеру, и поэтому повторная доставка не выполняется.
Попробуйте отключить обработку ошибок Camel, чтобы проверить, отправляет ли брокер сообщения об ошибках.
errorHandler(noErrorHandler());