@Bean
RabbitTemplate rabbitTemplate() {
RabbitTemplate template = new RabbitTemplate(rabbitConnectionFactory());
template.setMessageConverter(messageConverter);
template.setExchange(amqpProperties.getRabbitMqTopicExchangeName());
return template;
}
@Bean
@Conditional (OperationsCondition.class)
SimpleMessageListenerContainer opsMessageListenerContainer() {
return listenerContainer(amqpProperties.getRabbitMqOperationsQueue(),amqpProperties.getInitialRabbitOperationsConsumerCount(),amqpProperties.getMaximumRabbitOperationsConsumerCount(),opsReceiver());
}
@Bean
@Conditional (OperationsCondition.class)
OperationsListener opsReceiver() {
return new OperationsListener();
}
private SimpleMessageListenerContainer listenerContainer(String queue,int initConsumers,int maxConsumers,MessageListener listener)
{
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(rabbitConnectionFactory());
container.setQueueNames(queue);
container.setMessageListener(listener);
container.setConcurrentConsumers(initConsumers);
container.setMaxConcurrentConsumers(maxConsumers);
container.setMessageConverter(messageConverter);
return container;
}
消息监听器是:
public class OperationsListener implements MessageListener
{
public static final Logger logger = Logger.getInstance(OperationsListener.class);
@Autowired (required=true)
private OperationsProcessor processor;
@Autowired (required=true)
private ObjectMapper objectMapper;
public void onMessage(Message message)
{
Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter();
converter.setJsonObjectMapper(objectMapper);
OperationsMessage request = (OperationsMessage)converter.fromMessage(message);
processor.createMessage(request);
//This is throwing a JPA database exception
processor.createOperation(request);
}
}
processor.createOperation()因数据库问题而抛出异常.问题是消息监听器进入循环并且消息不断回来.
我的处理器类:
@Component
@Transactional (propagation = Propagation.REQUIRES_NEW)
public class OperationsProcessor
{
...............
public void createOperation(OperationsMessage message)
{
try
{
.............
.............
//this call throws exception.
opsRepo.create(operation,null);
}
catch (Exception e)
{
logger.error(e);
}
}
}
opsRepo.create抛出异常.即使我正在捕捉错误,我希望春天amqp不再发送该消息.不确定为什么同样的消息不断回来.
编辑:
我想我找到了一些关于如何处理这个问题的建议.原因是春天在失败时重新发生事件,这是默认的性质.
找到了一个有用的主题here和here.
最佳答案
原文链接:https://www.f2er.com/spring/431368.html