我将Spring与RabbitMQ结合使用,并且试图避免在发生运行时异常的情况下重新传递消息。我试图在requeue-reject
中将false
设置为listener-container
并配置抛出AmqpRejectAndDontRequeueException
的自定义错误处理程序。似乎这两种策略都失败了,并且消息永远继续传递。有什么想法的原因吗?
感谢帮助。
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit" xsi:schemaLocation=" http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.1.xsd">
<rabbit:connection-factory id="rabbitMQConnectionFactory" host="localhost" port="5672" username="guest" password="guest" />
<rabbit:admin connection-factory="rabbitMQConnectionFactory" />
<rabbit:template id="amqpTemplate" connection-factory="rabbitMQConnectionFactory" />
<rabbit:queue name="q1" />
<rabbit:queue name="q2" />
<rabbit:listener-container error-handler="errorHandler" connection-factory="rabbitMQConnectionFactory" concurrency="10" transaction-manager="transactionManager" requeue-rejected="false">
<rabbit:listener ref="q1Listener" method="consumeMessage" queue-names="q1" />
<rabbit:listener ref="q2Listener" method="consumeMessage" queue-names="q2" />
</rabbit:listener-container>
<bean id="errorHandler" class="ErrorHandler" />
<bean id="q1Listener" class="Q1MessageConsumerBean" />
<bean id="q2Listener" class="Q2MessageConsumerBean" />
</beans>
最佳答案
首先,对于目标侦听器中的requeue-rejected="false"
,AmqpRejectAndDontRequeueException
和RuntimeException
会执行相同的操作。
这是一个测试案例,表明requeue-rejected="false"
运作良好。
<rabbit:connection-factory id="connectionFactory" host="localhost" />
<rabbit:template id="amqpTemplate" connection-factory="connectionFactory"
exchange="foo" routing-key="foo" />
<rabbit:admin connection-factory="connectionFactory" />
<rabbit:queue name="foo" />
<rabbit:direct-exchange name="foo">
<rabbit:bindings>
<rabbit:binding queue="foo" key="foo" />
</rabbit:bindings>
</rabbit:direct-exchange>
<rabbit:listener-container connection-factory="connectionFactory" auto-startup="false" requeue-rejected="false">
<rabbit:listener ref="listener" queue-names="foo" />
</rabbit:listener-container>
<bean id="listener" class="org.mockito.Mockito" factory-method="spy">
<constructor-arg>
<bean class="org.springframework.amqp.rabbit.listener.RejectedTests$ThrowListener" />
</constructor-arg>
</bean>
@Autowired
private RabbitTemplate rabbitTemplate;
@Autowired
private SimpleMessageListenerContainer container;
@Autowired
private ThrowListener throwListener;
@Test
public void test() throws Exception {
rabbitTemplate.convertAndSend("foo");
container.start();
Thread.sleep(2000);
Mockito.verify(throwListener).onMessage(Mockito.any(Message.class));
}
public static class ThrowListener implements MessageListener {
@Override
public void onMessage(Message message) {
throw new RuntimeException("intentional reject");
}
}
Mockito.verify(throwListener).onMessage(Mockito.any(Message.class));
确认onMessage
在首次交付时仅被调用一次。由于我们的消息已被拒绝并从RabbitMQ Broker中删除,因此第二次交付没有发生。我只看到标志性的地方,它独立于
requeue-rejected="false"
-RabbitResourceHolder#rollbackAll()
不会发生:for (Long deliveryTag : deliveryTags.get(channel)) {
try {
channel.basicReject(deliveryTag, true);
} catch (IOException ex) {
throw new AmqpIOException(ex);
}
}
但是,仅在TX提交和该提交的Expception上到达此块。因此,在这种情况下,我们会导致TX回滚:
private boolean doReceiveAndExecute(BlockingQueueConsumer consumer) throws Throwable {
Channel channel = consumer.getChannel();
for (int i = 0; i < txSize; i++) {
logger.trace("Waiting for message from consumer.");
Message message = consumer.nextMessage(receiveTimeout);
if (message == null) {
break;
}
try {
executeListener(channel, message);
}
catch (ImmediateAcknowledgeAmqpException e) {
break;
}
catch (Throwable ex) {
consumer.rollbackOnExceptionIfNecessary(ex);
throw ex;
}
}
return consumer.commitIfNecessary(isChannelLocallyTransacted(channel));
注意最后一行。
consumer.rollbackOnExceptionIfNecessary(ex);
是从侦听器引发异常的情况。否则,我们到达最后一行并等待外部TX提交。让我们知道是否是您的情况。