問題描述
我正在嘗試使用 spring boot 創建一個簡單的 spring boot 應用程序，該應用程序向 rabbitmq 交換/隊列生成"消息，以及另一個使用"這些消息的示例 spring boot 應用程序.所以我有兩個應用程序(如果你愿意，也可以是微服務).1)生產者"微服務2)消費者"微服務
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
生產者"有 2 個域對象.Foo 和 Bar 應該被轉換為 json 并發送到 rabbitmq.消費者"應該接收 json 消息并將其分別轉換為域 Foo 和 Bar.由于某種原因，我無法完成這個簡單的任務.這方面的例子不多.對于消息轉換器，我想使用 org.springframework.messaging.converter.MappingJackson2MessageConverter
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.MappingJackson2MessageConverter
這是我目前所擁有的:
生產者微服務
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.MappingJackson2MessageConverter;
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 MappingJackson2MessageConverter jackson2Converter() {
MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
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 MappingJackson2MessageConverter mappingJackson2MessageConverter;
public void sendToRabbitmq(final Foo foo, final Bar bar) {
this.rabbitMessagingTemplate.setMessageConverter(this.mappingJackson2MessageConverter);
this.rabbitMessagingTemplate.convertAndSend("exchange", "queue", foo);
this.rabbitMessagingTemplate.convertAndSend("exchange", "queue", bar);
}
}
class Bar {
public int age = 33;
}
class Foo {
public String name = "gustavo";
}
消費者微服務
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;
}
這是我得到的例外:
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.access$001(SimpleMessageListenerContainer.java:83)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$1.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.access$700(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
異常說沒有轉換器，這是真的，我的問題是我不知道如何在消費者端設置MappingJackson2MessageConverter轉換器(請注意我要使用org.springframework.messaging.converter.MappingJackson2MessageConverter 而不是 org.springframework.amqp.support.converter.JsonMessageConverter)
The exception says there is no converter, and that is true, my problem is that I have no idea how to set the MappingJackson2MessageConverter converter in the consumer side (please note that I want to use org.springframework.messaging.converter.MappingJackson2MessageConverter and not org.springframework.amqp.support.converter.JsonMessageConverter)
有什么想法嗎?
以防萬一，您可以在以下位置創建此示例項目:https://github.com/gustavoorsi/rabbitmq-consumer-receiver
Just in case, you can fork this sample project at: https://github.com/gustavoorsi/rabbitmq-consumer-receiver
推薦答案
好的，我終于搞定了.
Spring 使用 PayloadArgumentResolver 來提取、轉換并將轉換后的消息設置為使用 @RabbitListener 注釋的方法參數.不知何故，我們需要將 ma??ppingJackson2MessageConverter 設置到這個對象中.
Spring uses a PayloadArgumentResolver to extract, convert and set the converted message to the method parameter annotated with @RabbitListener. Somehow we need to set the mappingJackson2MessageConverter into this object.
因此，在 CONSUMER 應用中，我們需要實現 RabbitListenerConfigurer.通過重寫 configureRabbitListeners(RabbitListenerEndpointRegistrar registrar) 我們可以設置一個自定義的 DefaultMessageHandlerMethodFactory，我們為這個工廠設置消息轉換器，工廠將創建我們的 PayloadArgumentResolver 使用正確的轉換.
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 PayloadArgumentResolver with the the correct convert.
這是代碼片段，我還更新了 git 項目.
Here is a snippet of the code, I've also updated the git project.
ConsumerApplication.java
package 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.MappingJackson2MessageConverter;
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 MappingJackson2MessageConverter jackson2Converter() {
MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
return converter;
}
@Bean
public DefaultMessageHandlerMethodFactory myHandlerMethodFactory() {
DefaultMessageHandlerMethodFactory factory = new DefaultMessageHandlerMethodFactory();
factory.setMessageConverter(jackson2Converter());
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;
}
因此，如果您運行 Producer 微服務，它將在隊列中添加 2 條消息.一個代表 Foo 對象，另一個代表 Bar 對象.通過運行消費者微服務，您將看到兩者都被 Receiver 類中的相應方法使用.
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 Receiver class.
更新問題:
我認為，我認為排隊存在概念上的問題.通過聲明 2 個用 @RabbitListener 注釋的指向同一個隊列的方法，我想要實現的目標是不可能的.上述解決方案無法正常工作.如果你發送到rabbitmq，比如說，6 條Foo 消息和3 條Bar 消息，它們不會被帶有Foo 參數的監聽器接收6 次.似乎偵聽器是并行調用的，因此無法根據方法參數類型區分要調用的偵聽器.我的解決方案(我不確定這是否是最好的方法，我在這里接受建議)是為每個實體創建一個隊列.所以現在，我有 queue.bar 和 queue.foo，并更新 @RabbitListener(queues = "queue.foo")我再次更新了代碼，您可以在我的 git 存儲庫中查看它.
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 @RabbitListener that 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.bar and 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.
這篇關于spring boot rabbitmq MappingJackson2MessageConverter 自定義對象轉換的文章就介紹到這了，希望我們推薦的答案對大家有所幫助，也希望大家多多支持html5模板網！