如何在中手动确认RabbitMQ消息？
如何在中手动确认RabbitMQ消息？
提问于 2018-01-10 23:26:04
对于基于流的服务，当在@StreamListener中调用的底层服务失败时，我希望消息保持在队列中。为此，我的理解是，实现此目的的唯一方法是配置spring.cloud.stream.bindings.channel_name.consumer.acknowledge-mode=MANUAL。
在进行此配置更改之后，我尝试将@Header(AmqpHeaders.CHANNEL) Channel channel,@Header(AmqpHeaders.DELIVERY_TAG) Long deliveryTag作为方法参数添加到现有的@StreamListener实现中，如在https://docs.spring.io/spring-integration/reference/html/amqp.html#amqp-inbound-ack上记录的那样。有了这段代码之后，我遇到了以下异常：
org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException: Listener threw exception
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.wrapToListenerExecutionFailedExceptionIfNeeded(AbstractMessageListenerContainer.java:941)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:851)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:771)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$001(SimpleMessageListenerContainer.java:102)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$1.invokeListener(SimpleMessageListenerContainer.java:198)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.invokeListener(SimpleMessageListenerContainer.java:1311)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:752)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:1254)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:1224)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:102)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1470)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.springframework.messaging.MessageHandlingException: Missing header 'amqp_channel' for method parameter type [interface com.rabbitmq.client.Channel]
at org.springframework.messaging.handler.annotation.support.HeaderMethodArgumentResolver.handleMissingValue(HeaderMethodArgumentResolver.java:100)
at org.springframework.messaging.handler.annotation.support.AbstractNamedValueMethodArgumentResolver.resolveArgument(AbstractNamedValueMethodArgumentResolver.java:103)
at org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite.resolveArgument(HandlerMethodArgumentResolverComposite.java:112)然后我发现了以下内容：示例，它展示了如何使用Kafka执行消息确认的示例，但我目前正在使用RabbitMQ绑定。我们计划最终转移到Kafka，但是现在，我如何配置并编写一个解决方案来对成功处理的消息和手动消息拒绝进行手动消息确认，从而在遇到异常时将消息留在队列中。我目前正在使用Spring Edgware.RELEASE和Stream Ditmars.RELEASE。
更新
现在我有了以下配置：
spring:
cloud:
stream:
bindings:
do-something-async-reply:
group: xyz-service-do-something-async-reply
rabbit:
bindings:
do-something-async-reply:
consumer:
autoBindDlq: true
dlqDeadLetterExchange:
dlqTtl: 10000
requeueRejected: true在服务启动时，我收到了以下错误：
2018-01-12 14:46:34.346 ERROR [xyz-service,,,] 2488 --- [ 127.0.0.1:5672] o.s.a.r.c.CachingConnectionFactory : Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'x-dead-letter-exchange' for queue 'do-something-async-reply.xyz-service-do-something-async-reply' in vhost '/': received the value 'DLX' of type 'longstr' but current is none, class-id=50, method-id=10)什么配置是错误的/我错过了吗？
回答 1
Stack Overflow用户
回答已采纳
发布于 2018-01-11 01:11:17
属性名不正确；您缺少.rabbit。它是
spring.cloud.stream.rabbit.bindings.
<channel>.consumer.acknowledge-mode=MANUAL
因为这是一个特定于兔子的属性-请参见文献资料。
编辑
示例：
@SpringBootApplication
@EnableBinding(Sink.class)
public class So481977082Application {
public static void main(String[] args) {
SpringApplication.run(So481977082Application.class, args);
}
@StreamListener(Sink.INPUT)
public void in(String in, @Header(AmqpHeaders.CHANNEL) Channel channel,
@Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception {
System.out.println(in);
Thread.sleep(60_000);
channel.basicAck(tag, false);
System.out.println("Ackd");
}
}请记住，对手动ack的需求通常是一种气味；通常情况下，让容器处理这些ack比较好；请参阅同一个doco链接上的requeueRejected。无条件请求会导致无限循环。
EDIT2
对我来说很好。
@SpringBootApplication
@EnableBinding(Processor.class)
public class So48197708Application {
public static void main(String[] args) {
SpringApplication.run(So48197708Application.class, args);
}
@Bean
ApplicationRunner runner(MessageChannel output) {
return args -> {
output.send(new GenericMessage<>("foo"));
};
}
@StreamListener(Sink.INPUT)
public void listen(@Header(name = "x-death", required = false) List<?> death) {
System.out.println(death);
throw new RuntimeException("x");
}
}使用
spring:
cloud:
stream:
bindings:
input:
group: foo
content-type: application/json
destination: foo
consumer:
max-attempts: 1
output:
content-type: application/json
destination: foo
rabbit:
bindings:
input:
consumer:
auto-bind-dlq: true
dlqDeadLetterExchange:
dlqTtl: 10000结果：
null
...
Caused by: java.lang.RuntimeException: x
...
[{reason=expired, count=1, exchange=DLX, routing-keys=[foo.foo], time=Fri Jan 12 17:20:28 EST 2018, queue=foo.foo.dlq},
{reason=rejected, count=1, exchange=foo, time=Fri Jan 12 17:20:18 EST 2018, routing-keys=[foo], queue=foo.foo}]
...
...
[{reason=expired, count=3, exchange=DLX, routing-keys=[foo.foo], time=Fri Jan 12 17:20:28 EST 2018, queue=foo.foo.dlq},
{reason=rejected, count=3, exchange=foo, routing-keys=[foo], time=Fri Jan 12 17:20:18 EST 2018, queue=foo.foo}]页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/48197708
复制相关文章
相似问题
腾讯云开发者
