问题标签 [spring-amqp]

我对 RabbitMQ 有一个有点独特的用例，但我不确定如何解决这个问题。我希望有一个绑定了多个消费者的队列，然后让 RabbitMQ 一次仅向一个消费者发送一条消息，并在向任何其他消费者发送另一条消息之前等待 ACK。

我意识到这会扼杀吞吐量，并且基本上会使其他消费者挨饿，但对我来说没关系。这个奇怪用例的原因是消费者与之交谈的服务一次只能处理一个并发请求，所以我需要一种方法来限制这一点，但消费者也可能意外死亡，我需要另一个消费者来接手处理消息如果发生这种情况。我知道有这个prefetch选项，但它仍然允许多个用户获得一个独占队列，但我不确定那些能完成我想要的。是否可以配置 RabbitMQ 来执行此操作？

RabbitMQ 新手和 Java 新手。

我正在尝试编写一个侦听器，它将使用手动确认并使用 java Spring AMQP 抽象处理消费者取消通知。我可以通过使用 Spring 抽象来完成这两项任务吗？

我想编写一个侦听器，它将从队列中提取消息并处理该消息（可能写入数据库或其他东西）。我计划使用手动确认，以便如果消息处理失败或由于某种原因无法完成，我可以拒绝并重新排队。到目前为止，我想我发现为了使用 Spring AMQP 手动确认/拒绝/拒绝，我必须使用ChannelAwareMessageListener.

我意识到我应该处理来自 RabbitMQ 的消费者取消通知，但是使用ChannelAwareMessageListener我并没有真正看到为此编写代码的方法。我看到处理 CCN 的唯一方法是使用较低级别的 java 客户端 api 编写代码，方法是调用和传递一个允许您处理消息传递和取消channel.basicConsume()的新实例。DefaultConsumer

我也看不到如何设置clientProperties（ConnectionFactory告诉代理我可以处理 CCN），因为我从 config.xml 中的 bean 获取工厂。

我的监听器和容器创建的伪代码如下。

我正在按照Stock Trading示例的说明进行操作，其中概述了如何使用请求/回复消息spring-amqp：http ://static.springsource.org/spring-amqp/docs/1.2.x/reference/html/sample-apps.html #d4e742

查看ServerHandler作为消息驱动的 POJO 的类，它恰好也能够发送回复消息......我没有看到可以注入 MessageProperties 的地方content-type

我已经调整了这个示例，我想返回一个伪装成简单字符串的 xml 消息......但是当ServerHandler我将回复发布到 RabbitMQ 时，它看起来像：

最终（我的猜测是）我的客户应该阅读这个......由于content_type设置不正确application/json并且我的用例失败而无法解析对象，所以我收到null回复作为回复。

Q) 有谁知道如何在 spring-amqp 请求/回复消息中设置回复的内容类型？

更新#1

问）也许我需要确保它jsonMessageConverter仅用于从 requestQueue 中读取消息，而不是在我想将回复放回 responseQueue 时使用......关于如何配置它的任何想法？这是我当前 appContext.xml 文件中的一个片段：

我正在遵循Stock Trading示例的说明，其中概述了如何使用请求/回复消息spring-amqp：http ://static.springsource.org/spring-amqp/docs/1.2.x/reference/html/sample-apps.html #d4e742

convertSendAndReceive我已经调整了示例说明以创建一个客户端，该客户端应该通过使用而不是等待回复convertAndSend：https ://gist.github.com/pulkitsinghal/5774487

现在，即使将回复放在 responseQueue 上，并且我已将超时更新为rabbitTemplate.setReplyTimeout(60000);比默认的 5 秒长……在我的客户端中，我也会收到null回复作为回复。

有谁知道发生了什么？

更新#1

有人建议我添加一个<reply-listener/>，<rabbit:template/>但我不确定如何以编程方式执行此操作：

我正在使用 Spring-AMQP 在某些服务之间进行通信。在服务内部，我声明了消费者。每个消费者使用一个队列来分派相应的消息。消费者经理拥有所有已知的消费者。实际的消息调度通过消息处理程序运行，该处理程序查找订阅相应消息类型的消费者。在几乎任何用例中，服务都只向一个消费者订阅消息。

对于特殊情况，我不仅要区分消息类型，还要区分消息来自哪里的队列。

是否可以以某种方式检索从中读取消息的队列名称？

我已经使用 POJO 通过 spring amqp 设置了基本的请求/回复消息，这些通过 JSON 消息转换器在 RabbitMQ 上交换。

到目前为止，我的响应 pojo 的有效负载是简单的字符串，但现在我有时也需要用图像回复。

Q1) 处理此问题的最佳方法是什么？图像是否应该被 base64 编码为 JSON 消息的有效负载字符串？

Q2）是否可以简单地将图像作为 BytesMessage 返回？

Q2.a) spring-amqp 是否能够处理两个单独的侦听器，一个返回 POJO，另一个返回 BytesMessage？我不这么认为，但这里有一些关于我要问的伪代码：

Q2.b) 或者...如果 MessageListener 返回一个对象（有时可能是 POJO，有时可能是 BytesMessage）... spring-amqp 框架是否能够在运行时进行相应调整以发送回序列化的 POJO有时作为 json 和其他时间的BytesMessage？

问题：我们有 2 或 3 个应用程序实例。每个实例都有一个生产者和一个消费者。我们必须安排一些流程，为此我们使用常见的 spring 调度程序。这个调度程序产生消息并将它们扔给“代理”（RabbitMQ）。在我们的例子中，我们处理相同的数据 2 或 3 次，因为每个实例都会抛出消息。在第一个生产者抛出消息之前，您将如何阻止实例的生产者？

配置：

在我的应用程序中，我有一个可能会变得非常大的队列。如果我发现我的机器上没有更多空间怎么办？如何在多台机器上拆分队列？也许 RabbitMQ 理念不同，我应该创建多个队列而不是一个大队列..？

最好的，弗拉维奥

我编写了示例 spring amqp 生产者，它在 RabbitMQ 服务器上运行，该服务器发送消息并使用 Spring AMQP 使用 MessageListener 使用这些消息。在这里，我想将队列和消息持久性设置为 false。您能否请任何人帮助我了解如何使用注释将“持久”标志设置为 false。

这是示例代码

提前致谢。

我正在使用 Spring AMQP 1.2 版本对 RabbitMQ 代理进行性能测试。任何人都可以解释相同的步骤并使用Java作为生产者和消费者的客户端。