Java spring boot rabbitmq MappingJackson2MessageConverter 自定义对象转换
声明：本页面是StackOverFlow热门问题的中英对照翻译，遵循CC BY-SA 4.0协议，如果您需要使用它，必须同样遵循CC BY-SA许可，注明原文地址和作者信息，同时你必须将它归于原作者（不是我）：StackOverFlow
原文地址: http://stackoverflow.com/questions/30770725/
Warning: these are provided under cc-by-sa 4.0 license. You are free to use/share it, But you must attribute it to the original authors (not me):
StackOverFlow
spring boot rabbitmq MappingHymanson2MessageConverter custom object conversion
提问by Gustavo
I'm trying to create a simple spring boot app with spring boot that "produce" messages to a rabbitmq exchange/queue and another sample spring boot app that "consume" these messages. So I have two apps (or microservices if you wish). 1) "producer" microservice 2) "consumer" microservice
我正在尝试使用 spring boot 创建一个简单的 spring boot 应用程序，该应用程序将消息“产生”到rabbitmq 交换/队列和另一个“消耗”这些消息的示例 spring boot 应用程序。所以我有两个应用程序（如果你愿意，也可以使用微服务）。1) “生产者”微服务 2) “消费者”微服务
The "producer" has 2 domain objects. Foo and Bar which should be converted to json and send to rabbitmq. The "consumer" should receive and convert the json message into a domain Foo and Bar respectively. For some reason I can not make this simple task. There are not much examples about this. For the message converter I want to use org.springframework.messaging.converter.MappingHymanson2MessageConverter
“生产者”有 2 个域对象。Foo 和 Bar 应该转换为 json 并发送到 rabbitmq。“消费者”应该接收 json 消息并将其分别转换为域 Foo 和 Bar。出于某种原因，我无法完成这个简单的任务。这方面的例子并不多。对于消息转换器，我想使用 org.springframework.messaging.converter.MappingHymanson2MessageConverter
Here is what I have so far:
这是我到目前为止所拥有的：
PRODUCER MICROSERVICE
生产者微服务
package demo.producer;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.core.RabbitMessagingTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.converter.MappingHymanson2MessageConverter;
import org.springframework.stereotype.Service;
@SpringBootApplication
public class ProducerApplication implements CommandLineRunner {
public static void main(String[] args) {
SpringApplication.run(ProducerApplication.class, args);
}
@Bean
Queue queue() {
return new Queue("queue", false);
}
@Bean
TopicExchange exchange() {
return new TopicExchange("exchange");
}
@Bean
Binding binding(Queue queue, TopicExchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with("queue");
}
@Bean
public MappingHymanson2MessageConverter Hymanson2Converter() {
MappingHymanson2MessageConverter converter = new MappingHymanson2MessageConverter();
return converter;
}
@Autowired
private Sender sender;
@Override
public void run(String... args) throws Exception {
sender.sendToRabbitmq(new Foo(), new Bar());
}
}
@Service
class Sender {
@Autowired
private RabbitMessagingTemplate rabbitMessagingTemplate;
@Autowired
private MappingHymanson2MessageConverter mappingHymanson2MessageConverter;
public void sendToRabbitmq(final Foo foo, final Bar bar) {
this.rabbitMessagingTemplate.setMessageConverter(this.mappingHymanson2MessageConverter);
this.rabbitMessagingTemplate.convertAndSend("exchange", "queue", foo);
this.rabbitMessagingTemplate.convertAndSend("exchange", "queue", bar);
}
}
class Bar {
public int age = 33;
}
class Foo {
public String name = "gustavo";
}
CONSUMER MICROSERVICE
消费者微服务
package demo.consumer;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.stereotype.Service;
@SpringBootApplication
@EnableRabbit
public class ConsumerApplication implements CommandLineRunner {
public static void main(String[] args) {
SpringApplication.run(ConsumerApplication.class, args);
}
@Autowired
private Receiver receiver;
@Override
public void run(String... args) throws Exception {
}
}
@Service
class Receiver {
@RabbitListener(queues = "queue")
public void receiveMessage(Foo foo) {
System.out.println("Received <" + foo.name + ">");
}
@RabbitListener(queues = "queue")
public void receiveMessage(Bar bar) {
System.out.println("Received <" + bar.age + ">");
}
}
class Foo {
public String name;
}
class Bar {
public int age;
}
And here is the exception I'm getting:
这是我得到的例外：
org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
Endpoint handler details:
Method [public void demo.consumer.Receiver.receiveMessage(demo.consumer.Bar)]
Bean [demo.consumer.Receiver@1672fe87]
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:116)
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.onMessage(MessagingMessageListenerAdapter.java:93)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:756)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:679)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.accesspackage demo.consumer;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.annotation.RabbitListenerConfigurer;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistrar;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.converter.MappingHymanson2MessageConverter;
import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;
import org.springframework.stereotype.Service;
@SpringBootApplication
@EnableRabbit
public class ConsumerApplication implements RabbitListenerConfigurer {
public static void main(String[] args) {
SpringApplication.run(ConsumerApplication.class, args);
}
@Bean
public MappingHymanson2MessageConverter Hymanson2Converter() {
MappingHymanson2MessageConverter converter = new MappingHymanson2MessageConverter();
return converter;
}
@Bean
public DefaultMessageHandlerMethodFactory myHandlerMethodFactory() {
DefaultMessageHandlerMethodFactory factory = new DefaultMessageHandlerMethodFactory();
factory.setMessageConverter(Hymanson2Converter());
return factory;
}
@Override
public void configureRabbitListeners(RabbitListenerEndpointRegistrar registrar) {
registrar.setMessageHandlerMethodFactory(myHandlerMethodFactory());
}
@Autowired
private Receiver receiver;
}
@Service
class Receiver {
@RabbitListener(queues = "queue")
public void receiveMessage(Foo foo) {
System.out.println("Received <" + foo.name + ">");
}
@RabbitListener(queues = "queue")
public void receiveMessage(Bar bar) {
System.out.println("Received <" + bar.age + ">");
}
}
class Foo {
public String name;
}
class Bar {
public int age;
}
1(SimpleMessageListenerContainer.java:83)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.invokeListener(SimpleMessageListenerContainer.java:170)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.invokeListener(SimpleMessageListenerContainer.java:1257)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:660)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:1021)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:1005)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access0(SimpleMessageListenerContainer.java:83)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1119)
at java.lang.Thread.run(Thread.java:745)
Caused by: org.springframework.amqp.support.converter.MessageConversionException: Cannot handle message
... 13 common frames omitted
Caused by: org.springframework.messaging.converter.MessageConversionException: No converter found to convert to class demo.consumer.Bar, message=GenericMessage [payload=byte[10], headers={amqp_receivedRoutingKey=queue, amqp_receivedExchange=exchange, amqp_deliveryTag=1, amqp_deliveryMode=PERSISTENT, amqp_consumerQueue=queue, amqp_redelivered=false, id=87cf7e06-a78a-ddc1-71f5-c55066b46b11, amqp_consumerTag=amq.ctag-msWSwB4bYGWVO2diWSAHlw, contentType=application/json;charset=UTF-8, timestamp=1433989934574}]
at org.springframework.messaging.handler.annotation.support.PayloadArgumentResolver.resolveArgument(PayloadArgumentResolver.java:115)
at org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite.resolveArgument(HandlerMethodArgumentResolverComposite.java:77)
at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.getMethodArgumentValues(InvocableHandlerMethod.java:127)
at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:100)
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:113)
... 12 common frames omitted
The exception says there is no converter, and that is true, my problem is that I have no idea how to set the MappingHymanson2MessageConverterconverter in the consumer side (please note that I want to use org.springframework.messaging.converter.MappingHymanson2MessageConverterand not org.springframework.amqp.support.converter.JsonMessageConverter)
异常说没有转换器，这是真的，我的问题是我不知道如何在消费者端设置MappingHymanson2MessageConverter转换器（请注意，我想使用org.springframework.messaging.converter.MappingHymanson2MessageConverter而不是org.springframework.amqp.support.converter.JsonMessageConverter)
Any thoughts ?
有什么想法吗 ？
Just in case, you can fork this sample project at: https://github.com/gustavoorsi/rabbitmq-consumer-receiver
以防万一，您可以在以下位置分叉此示例项目：https: //github.com/gustavoorsi/rabbitmq-consumer-receiver
采纳答案by Gustavo
Ok, I finally got this working.
好的，我终于得到了这个工作。
Spring uses a PayloadArgumentResolverto extract, convert and set the converted message to the method parameter annotated with @RabbitListener. Somehow we need to set the mappingHymanson2MessageConverterinto this object.
Spring 使用PayloadArgumentResolver提取、转换并将转换后的消息设置为使用@RabbitListener注释的方法参数。我们需要以某种方式将mappingHymanson2MessageConverter设置到这个对象中。
So, in the CONSUMER app, we need to implement RabbitListenerConfigurer. By overriding configureRabbitListeners(RabbitListenerEndpointRegistrar registrar)we can set a custom DefaultMessageHandlerMethodFactory, to this factory we set the message converter, and the factory will create our PayloadArgumentResolverwith the the correct convert.
因此，在 CONSUMER 应用程序中，我们需要实现RabbitListenerConfigurer。通过覆盖configureRabbitListeners(RabbitListenerEndpointRegistrar registrar)我们可以设置一个自定义DefaultMessageHandlerMethodFactory，我们为这个工厂设置消息转换器，工厂将使用正确的转换创建我们的PayloadArgumentResolver。
Here is a snippet of the code, I've also updated the git project.
这是代码片段，我还更新了git project。
ConsumerApplication.java
消费者应用程序
##代码##So, if you run the Producer microservice it will add 2 messages in the queue. One that represent a Foo object and another that represent a Bar object. By running the consumer microservice you will see that both are consumed by the respective method in the Receiverclass.
因此，如果您运行 Producer 微服务，它将在队列中添加 2 条消息。一个代表 Foo 对象，另一个代表 Bar 对象。通过运行消费者微服务，您将看到两者都由Receiver类中的相应方法使用。
Updated issue:
更新的问题：
There is a conceptual problem about queuing from my side I think. What I wanted to achieved can not be possible by declaring 2 methods annotated with @RabbitListenerthat points to the same queue. The solution above was not working properly. If you send to rabbitmq, let say, 6 Foo messages and 3 Bar messages, they wont be received 6 times by the listener with Foo parameter. It seems that the listener are invoked in parallel so there is no way to discriminate which listener to invoke based on the method argument type. My solution (and I'm not sure if this is the best way, I'm open to suggestions here) is to create a queue for each entity. So now, I have queue.barand queue.foo, and update @RabbitListener(queues = "queue.foo")Once again, I've updated the code and you can check it out in my git repository.
我认为从我这边排队有一个概念上的问题。通过声明 2 个用@RabbitListener注释的指向同一个队列的方法，我想要实现的目标是不可能的。上面的解决方案无法正常工作。如果您发送到 rabbitmq，例如 6 条 Foo 消息和 3 条 Bar 消息，带有 Foo 参数的侦听器将不会收到 6 次。似乎侦听器是并行调用的，因此无法根据方法参数类型区分要调用的侦听器。我的解决方案（我不确定这是否是最好的方法，我在这里接受建议）是为每个实体创建一个队列。所以现在，我有queue.bar和queue.foo，并更新@RabbitListener(queues = "queue.foo")我再次更新了代码，您可以在我的git 存储库中查看它。
回答by Rob Baily
Have not done this myself but it seems like you need to register the appropriate conversions by setting up a RabbitTemplate. Take a look at section 3.1.8 in this Spring documentation. I know it is configured using the AMQP classes but if the messaging class you are mentioning is compatible there is no reason you can't substitute it. Looks like this referenceexplains how you might do it using Java configuration rather than XML. I have not really used Rabbit so I don't have any personal experience but I would love to hear what you find out.
我自己没有这样做，但似乎您需要通过设置 RabbitTemplate 来注册适当的转换。请查看此 Spring 文档中的第 3.1.8 节。我知道它是使用 AMQP 类配置的，但是如果您提到的消息传递类兼容，则没有理由不能替换它。看起来这个参考解释了如何使用 Java 配置而不是 XML 来完成它。我没有真正使用过 Rabbit，所以我没有任何个人经验，但我很想听听您的发现。
