消费者之间的循环不工作？

Question

我正在尝试修改我的 spring-amqp 项目，以便在抛出某种异常类型时，始终将消息重新排队。否则，重试 x 次然后拒绝。

这里是相关的配置xml

<rabbit:connection-factory id="rabbitMqConnectionFactory" host="localhost" port="5672" />

<rabbit:template id="rabbitTemplate" connection-factory="rabbitMqConnectionFactory"
    exchange="my.exchange" routing-key="foo"/>

<bean id="myListenerContainer" class="org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer">
    <property name="connectionFactory" ref="rabbitMqConnectionFactory" />
    <property name="messageListener" ref="myMessageListener" />
    <property name="concurrentConsumers" value="2" />
    <property name="queueNames" value="work.q" />
    <property name="adviceChain" ref="myRetryInterceptor" />
</bean>

这里是相关代码

 @Bean(name="myRetryInterceptor")
public MethodInterceptor getInterceptor() {
    return RetryInterceptorBuilder.stateless()
        .retryPolicy(getRetryPolicy())
        .recoverer(new RejectAndDontRequeueRecoverer())
        .build();
}

private RetryPolicy getRetryPolicy() {      
    return new SimpleRetryPolicy(5, Collections.EMPTY_MAP) {
        @Override
        public boolean canRetry(RetryContext context) {
            Throwable t = context.getLastThrowable();
            if ((t!=null) && (t.getCause() instanceof com.test.MessageRetryException)) {
                return true;
            }
            return super.canRetry(context);
        }
    };
}

我注意到，当我故意在 myMessageListener 中抛出 com.test.MessageRetryException 时，同一线程会一遍又一遍地收到消息。这与没有修改弹簧类的实现形成对比。在这种情况下，消息交替地提供给一个消费者线程，然后是另一个。我做错了吗？

@Bean(name="myRetryInterceptor")
//all consumers are given the message here
public MethodInterceptor getInterceptor() {
    return RetryInterceptorBuilder.stateful()
            .maxAttempts(5)
            .recoverer(new RejectAndDontRequeueRecoverer())
            .build();
}

Answer 1

没有；你没有做错任何事；无状态重试在原始交付范围内完成（在同一个线程上）。在重试次数用尽之前不会发生消息拒绝。

另一方面，有状态重试会拒绝每次尝试的传递，它将被重新提交并（可能）由不同的线程处理。

由于 rabbitmq 现在将被拒绝的消息重新排入队列的头部，因此切换到有状态恢复并没有任何好处，它需要消息 id 标头才能工作（因此可以确定消息的状态）。

编辑：嗯 - 我看到你正在使用有状态重试；所以我希望重新交付发生在备用线程上。

如果您可以在 Gist 或某个地方发布 DEBUG 日志；我可以看看。但是，正如我所说，无状态重试无论如何都更简单。