问题标签 [spring-amqp]

我使用RabbitMQ和Spring Integration来处理传入的 JSON 消息。

配置的相关部分如下所示：

我使用 Jackson Databind 作为 JSON 转换器。

有时传入的 JSON 消息的语法不正确。这会导致以下（正确）异常：

然后由errorHandlerwhich 简单地处理异常MessagePublishingErrorHandlerto errorChannel。

到现在为止还挺好。问题是，该消息仍然被 AMQP 客户端拒绝，尽管我正在使用错误处理程序处理它。RabbitMQ 然后一遍又一遍地重新传递消息。即使配置死信队列也无济于事。任何想法如何正确处理这种情况？

处理过程中的异常（成功反序列化后）处理得很好：AMQP 消息确认和错误消息发送到errorChannel.

有任何想法吗？

库版本：

• 弹簧集成：3.0.1
• Spring框架：4.0.2
• 杰克逊数据绑定：2.3.1

我正在使用带有 Spring AMQP 抽象的 RabbitMQ。所以本质上我使用的是 Spring AMQP。

我需要处理连接失败。当您使用原始 RabbitMQ 类时，使用Lyra很容易实现这一点。

您如何在 Spring AMQP 中实现相同的功能？我希望我的代码不知道任何网络问题。

我知道 Spring 默认处理重新连接（以某种方式），但我想要的是 Lyra 风格的配置（无论是 XML 还是其他任何地方），所以我可以定义超时、最大重试次数、退避等。

我在 Spring-AMQP 中使用异步消息接收器来接收消息。目前只处理带有 JSON 内容的消息，但我还需要处理带有 XML 内容的消息。我当前的实现MessageListener有一个MessageConverter注入并在 中使用它onMessage(Message)，如下所示：

为了支持不同的内容类型，我显然可以使用MessageProperties来询问内容类型标题并手动选择要使用的转换器。但这似乎需要做很多工作，比如 Spring 应该为这种情况提供更好的支持。我希望找到一个通用的MessageConverter实现，可以从内容类型映射到特定的转换器，但似乎没有这样的事情。

编写这样的委派转换器是我的最佳选择吗？或者有没有办法配置ListenerContainer支持异步接收和根据需要自动使用的多个转换器？

我正在编写一个消息侦听器，它需要接受来自命名队列的消息。使用 spring-amqp，我想知道是否可以配置此侦听器，以便无论其支持交换类型是什么，都将通过队列消费消息。

例如：

这会将这个示例队列绑定到直接交换，但是如果消费者应该将队列绑定到直接、扇出、标头或主题交换怎么办？本质上，我想知道是否有一种方法可以将消费侦听器的队列绑定与在运行时必须知道其支持交换的确切类型分离。

我第一次尝试 RabbitMQ 并决定使用 Spring AMQP 包装器。我打算先测试写入性能并将发送者设置到队列中：

弹簧设置：

发件人主要：

但是，在发送 15 条消息后，客户端失败并出现以下异常：

怎么来的？

使用SimpleMessageListenerContainer附加到多个队列并配置有ChannelAwareMessageListener. 是否可以确定从哪个队列消费了一条消息？特别是如果消息是从 Exchange 路由到队列的。

看起来，如果消息直接发送到MessageProperties#getReceivedRoutingKey将包含队列名称的队列，但如果消息通过 Exchange 路由到队列，则此信息包含使用的路由键。

我正在寻找一种机制，无论消息如何传递到队列，都可以正确提取此信息。或者在 RabbitMQ 端使用包含此信息的标头来丰富信息的机制。

我正在使用 Spring AMQP 在 RabbitMQ 中创建队列。我想要一个队列，其名称包括运行应用程序的机器的名称。因此，队列名称可能是“fooQueue.host1”或“fooQueue.host2”，具体取决于您运行应用程序的位置。

我已经想出了一种方法来做到这一点（详见下文），但它似乎有点复杂。有没有更简单/更好/更弹簧的方法来完成这个？

我的解决方案

首先制作一个 bean 来获取机器名称：

然后在 Spring 配置中注册 bean：

然后在你的 Spring AMQP 配置中使用它，如下所示：

给定一个使用来自 RabbitMQ 队列的消息的基本MessageListener实现，我如何根据处理时可能引发的异常类型将消息发送到不同的死信队列？

最初发布消息的队列具有x-dead-letter-exchange并x-dead-letter-routing-key设置，但在我的情况下这还不够。

万一这很重要，我的应用程序正在使用 Spring 4 和 Spring Amqp。

I want to distribute the processing of large batches. The idea is to use Spring Batch fire up a bunch of AMQP consumers in a cloud and then to load cheap tasks (like item IDs) and submit them to an AMQP exchange. Writing of the results will be done by the consumers themselves.

Is there a ready-made library to accomplish this?

A few thoughts:

• Spring Batch is totally negotiable.
• Batch size is several millions. I don't want to kill my message broker by brute-force submitting all these IDs at once but use some kind of throttling instead.
• I do want to know about what items have been processed so I can monitor the process. So the batch controlling process will have to receive replies from the consumers.

我正在使用基于 spring rabbit mq 模板的侦听器

在负载增加的情况下，监听器只是简单地关闭

如何防止监听器关闭？

或者如何在关闭后立即重新启动监听器？

我正在使用以下配置：

消费者是我的听众，其中包含以下功能