如何使用标头交换从rabbitmq中的队列发送和接收java对象作为消息 [英] How to use header exchange to send and recieve java objects as messages from queues in rabbitmq
问题描述
我正在尝试使用 rabbitmq 中的标头交换来生成和使用消息.当我尝试生成消息时,我必须将消息转换为 byte[],但在消费者方面,我无法正确反序列化它.我在 jackson 依赖项中使用了 classMapper,但它仍然无法解码该类.我不确定是我遗漏了什么还是我做错了.
I am trying to use header exchange in rabbitmq to produce and consume messages. When i am trying to produce the message i have to convert the message to byte[] but on the consumer side i'm not able to deserialize it properly. I had used classMapper in my jackson dependency but still it is not able to decode the class. I am not sure if i'm missing something or if i'm doing it wrong.
ProducerConfiguration 类
ProducerConfiguration class
@Configuration
public class RabbitMQConfig
{
@Value("${exchangeName}")
String exchange;
@Value("${queueName}")
String queueName;
@Value("${queueName1}")
String queueName1;
@Bean
public List<Object> topicBindings()
{
Queue headerQueue = new Queue(queueName);
Queue headerQueue1 = new Queue(queueName1);
Map<String,Object> mapForQueue = new HashMap<>();
mapForQueue.put("key", "value");
mapForQueue.put("key2", "value2");
Map<String,Object> mapForQueue1 = new HashMap<>();
mapForQueue.put("key", "value");
mapForQueue1.put("key1", "value1");
HeadersExchange headerExchange = new HeadersExchange(exchange);
return Arrays.asList
(
headerQueue,headerQueue1,headerExchange,
BindingBuilder.bind(headerQueue).to(headerExchange).whereAny(mapForQueue),
BindingBuilder.bind(headerQueue1).to(headerExchange).whereAll(mapForQueue1)
);
}
@Bean
public Jackson2JsonMessageConverter jsonMessageConverter()
{
Jackson2JsonMessageConverter jsonConverter = new Jackson2JsonMessageConverter();
jsonConverter.setClassMapper(classMapper());
return jsonConverter;
}
@Bean
public DefaultClassMapper classMapper()
{
DefaultClassMapper classMapper = new DefaultClassMapper();
Map<String, Class<?>> idClassMapping = new HashMap<>();
idClassMapping.put("String", String.class);
idClassMapping.put("Employee", Employee.class);
classMapper.setIdClassMapping(idClassMapping);
return classMapper;
}
}
发送者类
@Service
public class RabbitMQSender
{
@Autowired
private AmqpTemplate amqpTemplate;
@Value("${exchangeName}")
private String exchange;
public void send(Employee employee)
{
Message message = MessageBuilder.withBody(SerializationUtils.serialize(employee))
.setHeader("key", "value")
.setHeader("key1", "value1")
.build();
amqpTemplate.convertAndSend(exchange,"",message);
}
}
消费者配置类
@EnableRabbit
@Configuration
public class ConsumerConfiguration
{
@Bean
public Jackson2JsonMessageConverter jsonMessageConverter()
{
Jackson2JsonMessageConverter jsonConverter = new Jackson2JsonMessageConverter();
jsonConverter.setClassMapper(classMapper());
return jsonConverter;
}
@Bean
public DefaultClassMapper classMapper()
{
DefaultClassMapper classMapper = new DefaultClassMapper();
Map<String, Class<?>> idClassMapping = new HashMap<>();
idClassMapping.put("String", String.class);
idClassMapping.put("Employee", Employee.class);
classMapper.setIdClassMapping(idClassMapping);
return classMapper;
}
}
消费者应用
@Component
@RabbitListener(queues = "#{'${combinedQueue}'.split(';')}")
public class RabbitMqListener
{
@RabbitHandler
public void recievedMessage(@Header(AmqpHeaders.CONSUMER_QUEUE) String rk,@Payload Employee employee)
{
System.out.println("*********************************************************************************************");
System.out.println("Header recieved = "+rk);
System.out.println("Recieved Object Message From Important Queue " + employee);
System.out.println("*********************************************************************************************");
}
@RabbitHandler
public void recievedMessage1(@Header(AmqpHeaders.CONSUMER_QUEUE) String rk,@Payload byte[] object)
{
System.out.println("*********************************************************************************************");
System.out.println("Header recieved = "+rk);
Employee emp = (Employee)SerializationUtils.deserialize(object);
System.out.println("Recieved String Message From Important Queue " + emp);
System.out.println("*********************************************************************************************");
}
}
员工类
public class Employee
{
private String empName;
private String empId;
public String getEmpName() {
return empName;
}
public void setEmpName(String empName) {
this.empName = empName;
}
public String getEmpId() {
return empId;
}
public void setEmpId(String empId) {
this.empId = empId;
}
@Override
public String toString() {
return "Employee [empName=" + empName + ", empId=" + empId + "]";
}
}
错误日志
org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException: Listener method 'public void com.rabbitmq.consumer.service.RabbitMqListener.recievedMessage1(java.lang.String,byte[])' threw exception
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:198) ~[spring-rabbit-2.1.5.RELEASE.jar:2.1.5.RELEASE]
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.onMessage(MessagingMessageListenerAdapter.java:127) ~[spring-rabbit-2.1.5.RELEASE.jar:2.1.5.RELEASE]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1552) ~[spring-rabbit-2.1.5.RELEASE.jar:2.1.5.RELEASE]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1478) ~[spring-rabbit-2.1.5.RELEASE.jar:2.1.5.RELEASE]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1466) ~[spring-rabbit-2.1.5.RELEASE.jar:2.1.5.RELEASE]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1461) ~[spring-rabbit-2.1.5.RELEASE.jar:2.1.5.RELEASE]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1410) ~[spring-rabbit-2.1.5.RELEASE.jar:2.1.5.RELEASE]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:870) [spring-rabbit-2.1.5.RELEASE.jar:2.1.5.RELEASE]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:854) [spring-rabbit-2.1.5.RELEASE.jar:2.1.5.RELEASE]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:78) [spring-rabbit-2.1.5.RELEASE.jar:2.1.5.RELEASE]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1137) [spring-rabbit-2.1.5.RELEASE.jar:2.1.5.RELEASE]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1043) [spring-rabbit-2.1.5.RELEASE.jar:2.1.5.RELEASE]
at java.lang.Thread.run(Thread.java:748) [na:1.8.0_191]
Caused by: java.lang.IllegalStateException: Could not deserialize object type
at org.springframework.amqp.utils.SerializationUtils.deserialize(SerializationUtils.java:98) ~[spring-amqp-2.1.5.RELEASE.jar:2.1.5.RELEASE]
at org.springframework.amqp.utils.SerializationUtils.deserialize(SerializationUtils.java:75) ~[spring-amqp-2.1.5.RELEASE.jar:2.1.5.RELEASE]
at com.rabbitmq.consumer.service.RabbitMqListener.recievedMessage1(RabbitMqListener.java:32) ~[classes/:na]
at sun.reflect.GeneratedMethodAccessor42.invoke(Unknown Source) ~[na:na]
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_191]
at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_191]
at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:170) ~[spring-messaging-5.1.6.RELEASE.jar:5.1.6.RELEASE]
at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:120) ~[spring-messaging-5.1.6.RELEASE.jar:5.1.6.RELEASE]
at org.springframework.amqp.rabbit.listener.adapter.DelegatingInvocableHandler.invoke(DelegatingInvocableHandler.java:130) ~[spring-rabbit-2.1.5.RELEASE.jar:2.1.5.RELEASE]
at org.springframework.amqp.rabbit.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:60) ~[spring-rabbit-2.1.5.RELEASE.jar:2.1.5.RELEASE]
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:190) ~[spring-rabbit-2.1.5.RELEASE.jar:2.1.5.RELEASE]
... 12 common frames omitted
Caused by: java.lang.ClassNotFoundException: com.poc.springbootrabbitmq.model.Employee
at java.net.URLClassLoader.findClass(URLClassLoader.java:382) ~[na:1.8.0_191]
at java.lang.ClassLoader.loadClass(ClassLoader.java:424) ~[na:1.8.0_191]
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:349) ~[na:1.8.0_191]
at java.lang.ClassLoader.loadClass(ClassLoader.java:357) ~[na:1.8.0_191]
at java.lang.Class.forName0(Native Method) ~[na:1.8.0_191]
at java.lang.Class.forName(Class.java:348) ~[na:1.8.0_191]
at java.io.ObjectInputStream.resolveClass(ObjectInputStream.java:686) ~[na:1.8.0_191]
at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1868) ~[na:1.8.0_191]
at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1751) ~[na:1.8.0_191]
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2042) ~[na:1.8.0_191]
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1573) ~[na:1.8.0_191]
at java.io.ObjectInputStream.readObject(ObjectInputStream.java:431) ~[na:1.8.0_191]
at org.springframework.amqp.utils.SerializationUtils.deserialize(SerializationUtils.java:92) ~[spring-amqp-2.1.5.RELEASE.jar:2.1.5.RELEASE]
... 22 common frames omitted
2019-05-06 00:39:35.801 WARN 2780 --- [ntContainer#0-1] o.s.a.s.c.Jackson2JsonMessageConverter : Could not convert incoming message with content-type [application/octet-stream], 'json' keyword missing.
另外,我可以生成除 byte[] 以外的任何格式的消息吗???
Also, can i produce the message in any format other than byte[] ???
推荐答案
2019-05-06 00:39:35.801 WARN 2780 --- [ntContainer#0-1] osascJackson2JsonMessageConverter:无法转换内容类型为 [application/octet-stream] 的传入消息,缺少json"关键字.
2019-05-06 00:39:35.801 WARN 2780 --- [ntContainer#0-1] o.s.a.s.c.Jackson2JsonMessageConverter : Could not convert incoming message with content-type [application/octet-stream], 'json' keyword missing.
Jackson 转换器仅在 content_type
属性包含 json
(例如 application/json
)时才会进行转换.
The Jackson converter will only convert if the content_type
property contains json
(e.g. application/json
).
这篇关于如何使用标头交换从rabbitmq中的队列发送和接收java对象作为消息的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!