问题标签 [spring-integration-amqp]

我想处理一条 AMQP 消息，然后将其发送到另一个队列以进行进一步处理。

我正在使用 Spring 集成 DSL 将其存档，如下所示，

但是我发现当新消息在类型处理阶段处理时， NOTE_INCOMING_QUEUE队列中的消息仍然未被确认。请参阅下面的 rabbitmq 管理截图。

我想知道为什么即使处理程序已经成功执行， NOTE_INCOMING_QUEUE中的消息仍然未被确认。是spring集成amqp的设计还是我的代码有问题？

我正在通过 spring-integration 使用 spring-amqp。
对了，在使用spring-cloud-sleuth-zipkin的时候，出现了如下错误。

spring-cloud-starter-zipkin:1.0.9
spring-integration-core:4.2.9
spring-amqp:1.5.6

目前我正在尝试使用 json 格式将消息放入队列。下面是我的代码片段，但它不起作用。

该消息作为application/x-java-serialized-object放入队列中。

我有一些通过调用 DirectExchange 来调用 Spring Integration 的客户端代码，例如

我可以看到调用了集成流，组装了正确的结果，然后将 AMQP 消息发送回，但从未收到该消息。我必须使用 XML 配置，并且我尝试了通道适配器和网关的各种组合，取得了不同程度的成功，但没有任何组合始终返回。

只是一个入站网关会调用流，客户端代码会看到响应，但一半的响应是输入本身：

通道适配器也调用流，但客户端代码从不使用响应，无论是否使用交换和路由表达式，尽管 AMQP 消息似乎已组装和发布。

使用这两种方法，我缺少什么配置步骤？

编辑：完整示例上下文；使用入站网关的响应时间减半；使用通道适配器不会。无论我一个接一个地使用链或单个通道，行为都是相同的。

在测试中快速而肮脏地应用客户端代码：

我有一个使用 Spring Cloud Stream Rabbit 和 Eureka Discovery Client 的简单 Spring Boot 应用程序。该应用程序可以与 Eureka Server 一起正常工作，并且通过 RabbitMQ 的消息传递也可以正常工作。但是如果我启动一个 Spring Boot Admin Server，应用程序会开始记录健康检查失败：

Spring Boot Admin 仪表板也将应用程序显示为 DOWN，因为“Rabbit”运行状况指示器为 DOWN。

pom.xml：

带有代码的 GitHub 存储库以重现

我有一个IntegrationFlow这样定义的弹簧集成：

serviceActivatorBean看起来像这样：

如果myMethod抛出异常，它将被记录但不会重试。我试图将其更改IntegrationFlow为：

但后来我收到这样的日志消息（不会发生重试）：

2017-06-30 13:18:10.611 WARN 88706 --- [erContainer#1-2] osihaRequestHandlerRetryAdvice：此建议 org.springframework.integration.handler.advice.RequestHandlerRetryAdvice 只能用于 MessageHandlers；在“org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$1”中建议方法“invokeListener”的尝试被忽略

我怎样才能将其配置IntegrationFlow为与 a 一样的行为方式RabbitListener？即让RabbitMQ 再次发布消息。

我已经声明了具有单个输入和输出通道的 int:chain，

如何声明多个输出通道，每个通道具有不同的转换方法（不同的消息）

提前致谢

我正在将消息发布到 RabbitMQ 并且我想在 RabbitMQ 关闭时跟踪错误，为此我添加了一个RetryTemplate带有恢复回调的消息，但恢复回调仅提供此方法getLastThrowable()，我不确定如何提供详细信息RabbitMQ 关闭时失败的消息。（根据文档“RecoveryCallback在重试上下文仅包含该 lastThrowable字段时有所限制。对于更复杂的用例，您应该使用外部 RetryTemplate，以便您可以RecoveryCallback通过上下文的属性将附加信息传达给知道如何做到这一点，如果有人可以帮助我举一个很棒的例子。

兔子模板

恢复回调

AMQP 出站适配器

return IntegrationFlows.from("eventsChannel") .split() .handle(Amqp.outboundAdapter(rabbitMqTemplate) .exchangeName(exchangeName) .confirmCorrelationExpression("payload") .confirmAckChannel(ackChannel) .confirmNackChannel(nackChannel) ) .get();

我目前正在阅读 spring 集成主题。在spring-integration的XML配置中遇到了一个符号/前缀->“int”。

任何人都可以帮助我理解它吗？我也在某个地方看到使用了“si”。这些前缀究竟指定了什么&它们是特定于弹簧集成的吗？提前致谢！

我正在使用 RabbitMQ 3.6.10 UI 发布一条消息，该消息由使用 Spring Integration AMQP 4.3.11 的 Java 应用程序接收。该消息是对使用拆分器创建的较早消息的回复，因此它具有 asequenceNumber和sequenceSize标头。我将这些标头复制到回复中，并将它们设置为NumberRabbitMQ UI 中的类型。但是，在 Java 方面，我遇到了一个异常：

我检查了 Java 端的sequenceNumber和标头的类型是，而不是。然而，在 RabbitMQ UI 中没有选项可以做出这种改变。消息将由非 Java 应用程序发送，那么如何确保标头被 Spring Integration 识别为整数？sequenceSizeLongInteger

当我使用 Java 客户端发布回复并将标头值设置为 时Integer，消费者会接受它们。所以这可能是 RabbitMQ UI 没有足够具体的标头类型（例如 32 位与 64 位数字）或 Java 客户端对预期值类型过于严格的限制。任何人都可以确认其中一个吗？