问题标签 [spring-amqp]

我想根据消息头以不同的方式处理传入的消息。如何在 RabbitMQ 中高效实现？

我的听众正在收听来自 present.queue 频道的消息。消息头 KEY 有不同的键：key1，key2。带有 key1 的消息应该由 messageService.method1 处理，带有 key2 的消息应该由 messageService.method2 处理。

这是我正在使用的代码，但显然它是错误的。我可能应该将消息从 present.queue 路由到 first.queue 和 second.queue ？

我在 java 中使用 amqp-client 但 JVM 在关闭通道时无限挂起。
如果我删除channel.close()它 jvm 会无限挂断connection.close()。
我浏览了 API 类，发现在这两种情况下，RabbitMQ API 都将超时视为无限，它只是等待回复。

请告诉他们是否有任何解决方法。我正在使用 amqp-client-3.1.3。
非常感谢。

我们将在我们的项目中使用rabbitmq，但面临一个问题，我们想在我们的开发机器上调试，所以必须将响应消息发送到最初发送请求消息的机器。我们将如何实现这一目标，spring-rabbitmq 框架中是否有现有的解决方案？

我们考虑了几种解决方案。比如为每台机器声明一组队列，队列名前缀为机器名。这可行吗？

我有一个 Spring MVC 应用程序。我想使用 spring ampq 来执行我的异步任务。我的 REST 后端使用 @Transactional 注释来管理事务。完成工作后，它会将任务推送到交易所，然后返回。然后任务被消费者接收，它在第一次尝试时收到“org.hibernate.HibernateException: No Session found for current thread”错误。然后 spring-ampq 重试将任务发送给消费者，这一次它可以工作，没有任何会话错误。

我怎样才能使这种情况在第一次尝试中起作用？

我的 spring ampq 配置与此类似（http://projects.spring.io/spring-amqp/）：

注意： foo.Foo listen() 方法也用@Transactional 注释。

我的事务管理器配置如下：

注意：会话工厂是在 dao 工厂深处初始化的。我在 REST 服务和兔子消费者中使用相同的 dao 工厂。

我的错误日志如下：

监听代码如下：

REST 后端的代码与此类似：

我正在支持测试Spring-AMQP，Spring-Integration我已经进行了配置和测试：

在此配置中，很明显，一旦消息到达，consumingChannel它们就会被确认并因此从队列中删除。sleep我通过在后面放高receive并检查来验证这一点queue-size。没有进一步的控制。

现在，如果我设置acknowledge-mode=MANUAL，似乎没有办法ack通过弹簧集成进行手动操作。

我需要处理消息，并在处理后执行此操作manual-ack，直到ack消息保持在durableQ.

有什么方法可以处理MANUALackspring-amqp-integration吗？我想避免传递ChannelAwareMessageListener给，inbound-channel-adapter因为我想控制消费者的receive.

更新：

listener-container使用 own with时，这似乎是不可能的inbound-channel-adapter：

messageListener由于不允许使用属性，上述配置会引发错误，请参阅标记的内联注释。所以使用的目的listner-container被打败了（暴露channelvia ChannelAwareMessageListener）。

对我来说spring-integration不能用于manual-acknowledgement（我知道，这是一个很难说的说法！），任何人都可以帮助我验证这一点，或者我是否缺少任何特定的方法/配置？

谁能告诉我有什么区别

org.springframework.amqp.support.converter.JsonMessageConverter

和

org.springframework.http.converter.json.MappingJacksonHttpMessageConverter?

我目前正在使用带有 spring (spring-rabbit-1.2.0-RELEASE) 的 RabbitMQ，配置如下：

我想在第三次尝试后将消息重新排入队列尾部。但我找不到执行此操作的方法。

有人有想法吗？

在此先感谢您的帮助。

我正在使用 Spring Integration & SI AMQP 3.0.0-RELEASE。

我在两个 SI 实例之间通过 AMQP 有一个相当简单的请求-响应。

我发现当响应返回到请求服务器时，SI 正在尝试使用 Request 对象的类型而不是 Response 对象来反序列化响应。

即，给定网关接口：

我发现即使 an 的正确 JSONAnalyticsResponse到达服务器，SI 仍试图将其反序列化为EntityMessage，但失败了。

我已经调试过了，我怀疑原因是响应方正在复制入站json__TypeId__标头，而不是提供自己的标头。但是，我看不到我在哪里配置错误。

这是我的配置——我做错了什么？

请求方：

响应方：

Spring AMQP：比较多个消费者与每个消费者的多个线程的性能

我正处于从 Spring doc 学习 Spring AMQP 的阶段。我不清楚增加消费率异步消息的首选方法：根据 Spring 文档（http://docs.spring.io/spring-amqp/docs/1.2.1.RELEASE/reference/html/amqp.html#d4e329 ) ，以下两点用于配置 SimpleMessageListenerContainer：

1. concurrency => 为每个侦听器启动的并发消费者数量。在增加计数时，这将增加队列中接收的消费者数量。

2. taskExecutor => 默认情况下，这确保为每个传入消息创建新线程

所以，我假设如果我们有无限线程来执行任务（第 2 点），那么就没有必要增加消费者的数量。请求验证我的假设并发送您记录的任何性能点

我有以下 xml，我想用@Configuration类替换这个烦人的 xml

和我的网关：

我已经在为 spring-amqp 使用配置注释类，它工作得很好。我怎样才能对 Spring Integration 配置做同样的事情？

PS：我正在使用 Spring 3.2 和 Spring integration 3.0