我正在尝试创建一个简单的 spring boot 应用程序，其中 spring boot 将消息“生成”到 rabbitmq 交换/队列，另一个示例 spring boot 应用程序“使用”这些消息。 所以我有两个应用程序(如果你愿意，也可以是微服务)。 1)“生产者”微服务 2)“消费者”微服务
“生产者”有 2 个域对象。 Foo 和 Bar 应转换为 json 并发送到 rabbitmq。 “消费者”应该接收 json 消息并将其分别转换为域 Foo 和 Bar。 出于某种原因，我无法完成这个简单的任务。这方面的例子不多。 对于消息转换器，我想使用 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 转换器(请注意我想使用 < strong="">org.springframework.messaging.converter.MappingJackson2MessageConverter 而不是 org.springframework.amqp.support.converter.JsonMessageConverter)
有什么想法吗？
以防万一，您可以在以下位置创建示例项目: https://github.com/gustavoorsi/rabbitmq-consumer-receiver
最佳答案
好吧，我终于搞定了。
Spring 使用PayloadArgumentResolver 提取、转换并将转换后的消息设置为用@RabbitListener 注释的方法参数。我们需要以某种方式将 mappingJackson2MessageConverter 设置到该对象中。
因此，在 CONSUMER 应用中，我们需要实现 RabbitListenerConfigurer。通过覆盖configureRabbitListeners(RabbitListenerEndpointRegistrar registrar)，我们可以设置一个自定义的DefaultMessageHandlerMethodFactory，我们为这个工厂设置消息转换器，工厂将创建我们的PayloadArgumentResolver 使用正确的转换。
这是代码片段，我还更新了 git project .
消费者应用程序.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 类中的相应方法使用。
更新问题:
我认为在我这边排队存在概念性问题。通过声明 2 个用 @RabbitListener 注释的指向同一队列的方法，我无法实现我想要实现的目标。上面的解决方案无法正常工作。如果您向 rabbitmq 发送 6 条 Foo 消息和 3 条 Bar 消息，它们将不会被带有 Foo 参数的监听器接收 6 次。似乎监听器是并行调用的，因此无法根据方法参数类型来区分调用哪个监听器。 我的解决方案(我不确定这是否是最好的方法，我在这里接受建议)是为每个实体创建一个队列。 所以现在，我有 queue.bar 和 queue.foo，并更新 @RabbitListener(queues = "queue.foo") 我再次更新了代码，您可以在我的 git repository 中查看.
关于java - spring boot rabbitmq MappingJackson2MessageConverter 自定义对象转换，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/30770725/