ThinkChat🤖让你学习和工作更高效,注册即送10W Token,即刻开启你的AI之旅 广告
[TOC] # Jackson2JsonMessageConverter ## Json MessageConverter 先看一个demo ### 消费端: ~~~java import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter; import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class MQConfig { @Bean public ConnectionFactory connectionFactory(){ CachingConnectionFactory factory = new CachingConnectionFactory(); factory.setUri("amqp://zhihao.miao:123456@192.168.1.131:5672"); return factory; } @Bean public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory){ RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory); return rabbitAdmin; } @Bean public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory){ RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory); return rabbitTemplate; } @Bean public SimpleMessageListenerContainer messageListenerContainer(ConnectionFactory connectionFactory){ SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); container.setConnectionFactory(connectionFactory); container.setQueueNames("zhihao.miao.order"); MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageHandler()); //指定Json转换器 adapter.setMessageConverter(new Jackson2JsonMessageConverter()); //设置处理器的消费消息的默认方法 adapter.setDefaultListenerMethod("onMessage"); container.setMessageListener(adapter); return container; } } ~~~ 消息转换器使用了RabbitMQ自带的`Jackson2JsonMessageConverter`转换器,但是没有指定消息的`contentType`类型 处理器,定义了二个消息处理方法,参数不一样: ~~~ public class MessageHandler { public void onMessage(byte[] message){ System.out.println("---------onMessage----byte-------------"); System.out.println(new String(message)); } public void onMessage(String message){ System.out.println("---------onMessage---String-------------"); System.out.println(message); } ~~~ 消费端应用启动类: ~~~java import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.ComponentScan; import java.util.concurrent.TimeUnit; @ComponentScan public class Application { public static void main(String[] args) throws Exception{ AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(Application.class); System.out.println("===start up======"); TimeUnit.SECONDS.sleep(60); context.close(); } } ~~~ ### 生产端代码 ~~~java import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class MQConfig { @Bean public ConnectionFactory connectionFactory(){ CachingConnectionFactory factory = new CachingConnectionFactory(); factory.setUri("amqp://zhihao.miao:123456@192.168.1.131:5672"); return factory; } @Bean public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory){ RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory); return rabbitAdmin; } @Bean public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory){ RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory); return rabbitTemplate; } } ~~~ 消息的实体类型 ~~~java @Data public class Order { private Integer id; private Integer userId; private double amout; private String time; } ~~~ 应用启动类,生产端传递的消息类型是Order类型,并且转换成JSON类型发送到队列中 ~~~java import com.fasterxml.jackson.databind.ObjectMapper; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.ComponentScan; import java.time.LocalDateTime; @ComponentScan public class Application { public static void main(String[] args) throws Exception{ AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(Application.class); RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class); System.out.println(rabbitTemplate); Order order = new Order(); order.setId(1); order.setUserId(1000); order.setAmout(88d); order.setTime(LocalDateTime.now().toString()); ObjectMapper mapper = new ObjectMapper(); String json = mapper.writeValueAsString(order); System.out.println(json); rabbitTemplate.convertAndSend("","zhihao.miao.order",json); context.close(); } } ~~~ 消费之后的控制台打印: ~~~= ===start up====== ---------onMessage----byte------------- 九月 08, 2017 10:25:20 下午 org.springframework.amqp.support.converter.Jackson2JsonMessageConverter fromMessage 警告: Could not convert incoming message with content-type [text/plain] {"id":1,"userId":1000,"amout":88.0,"time":"2017-09-08T22:03:46.015"} ~~~ 我们发现消费端还是将其当作字节数组来消费,转换器还是将其转换成byte\[\] ### 改造 此时是因为生产端没有指定contentType类型,生产者应用启动类重新指定了相应的contentType类型后, ~~~java import com.fasterxml.jackson.databind.ObjectMapper; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.ComponentScan; import java.time.LocalDateTime; @ComponentScan public class Application { public static void main(String[] args) throws Exception{ AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(Application.class); RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class); System.out.println(rabbitTemplate); Order order = new Order(); order.setId(1); order.setUserId(1000); order.setAmout(88d); order.setTime(LocalDateTime.now().toString()); ObjectMapper mapper = new ObjectMapper(); String json = mapper.writeValueAsString(order); System.out.println(json); MessageProperties messageProperties = new MessageProperties(); messageProperties.setContentType("application/json"); Message message = new Message(json.getBytes(),messageProperties); rabbitTemplate.send("","zhihao.miao.order",message); context.close(); } } ~~~ 此时消费端的`Jackson2JsonMessageConverter`类型转换器将其转换成Map类型,指定消费的方法参数类型是Map即可。 ~~~ public class MessageHandler { public void onMessage(byte[] message){ System.out.println("---------onMessage----byte-------------"); System.out.println(new String(message)); } public void onMessage(String message){ System.out.println("---------onMessage---String-------------"); System.out.println(message); } public void onMessage(Map order){ System.out.println("---------onMessage---map-------------"); System.out.println(order.toString()); } } ~~~ 此时消费端控制台打印,我们知道生产者传递JSON类型数据,消费者将其作为Map类型的数据进行处理: ~~~ ---------onMessage---map------------- {id=1, userId=1000, amout=88.0, time=2017-10-15T22:47:03.500} ~~~ ### 再次改造 如果消费端发送多条消息,发送List的json格式,那么在消费端也要使用参数是List的方法来消费,生产者启动应用类 ~~~java import com.fasterxml.jackson.databind.ObjectMapper; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.ComponentScan; import java.time.LocalDateTime; import java.util.ArrayList; import java.util.List; @ComponentScan public class Application { public static void main(String[] args) throws Exception{ AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(Application.class); RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class); System.out.println(rabbitTemplate); Order order = new Order(); order.setId(1); order.setUserId(1000); order.setAmout(88d); order.setTime(LocalDateTime.now().toString()); ObjectMapper mapper = new ObjectMapper(); String json = mapper.writeValueAsString(order); System.out.println(json); MessageProperties messageProperties = new MessageProperties(); messageProperties.setContentType("application/json"); Message message = new Message(json.getBytes(),messageProperties); rabbitTemplate.send("","zhihao.miao.order",message); Order order2 = new Order(); order2.setId(2); order2.setUserId(2000); order2.setAmout(99d); order2.setTime(LocalDateTime.now().toString()); List<Order> orderList = new ArrayList<>(); orderList.add(order); orderList.add(order2); String jsonlist = mapper.writeValueAsString(orderList); Message message2 = new Message(jsonlist.getBytes(),messageProperties); rabbitTemplate.send("","zhihao.miao.order",message2); context.close(); } } ~~~ 消费端的Handler: ~~~ public class MessageHandler { public void onMessage(byte[] message){ System.out.println("---------onMessage----byte-------------"); System.out.println(new String(message)); } public void onMessage(String message){ System.out.println("---------onMessage---String-------------"); System.out.println(message); } public void onMessage(Map order){ System.out.println("---------onMessage---map-------------"); System.out.println(order.toString()); } public void onMessage(List orders){ System.out.println("---------onMessage---List-------------"); System.out.println(orders.toString()); } } ~~~ 消费者控制台打印,此时发现消费端将消息转换成List类型的消息: ~~~ ---------onMessage---map------------- {id=1, userId=1000, amout=88.0, time=2017-10-15T22:52:46.739} ---------onMessage---List------------- [{id=1, userId=1000, amout=88.0, time=2017-10-15T22:52:46.739}, {id=2, userId=2000, amout=99.0, time=2017-10-15T22:52:47.882}] ~~~ ### 总结 * 使用Jackson2JsonMessageConverter处理器,客户端发送JSON类型数据,但是没有指定消息的contentType类型,那么Jackson2JsonMessageConverter就会将消息转换成byte\[\]类型的消息进行消费。 * 如果指定了contentType为application/json,那么消费端就会将消息转换成Map类型的消息进行消费。 * 如果指定了contentType为application/json,并且生产端是List类型的JSON格式,那么消费端就会将消息转换成List类型的消息进行消费。 ## Jackson2JsonMessageConverter类的源码分析 ~~~ @Override public Object fromMessage(Message message) throws MessageConversionException { Object content = null; MessageProperties properties = message.getMessageProperties(); if (properties != null) { String contentType = properties.getContentType(); //contentType中包含有json的都是用指定的格式来转换消息 if (contentType != null && contentType.contains("json")) { String encoding = properties.getContentEncoding(); if (encoding == null) { encoding = getDefaultCharset(); } try { if (getClassMapper() == null) { JavaType targetJavaType = getJavaTypeMapper() .toJavaType(message.getMessageProperties()); content = convertBytesToObject(message.getBody(), encoding, targetJavaType); } else { Class<?> targetClass = getClassMapper().toClass( message.getMessageProperties()); content = convertBytesToObject(message.getBody(), encoding, targetClass); } } catch (IOException e) { throw new MessageConversionException( "Failed to convert Message content", e); } } else { if (log.isWarnEnabled()) { log.warn("Could not convert incoming message with content-type [" + contentType + "]"); } } } //其余的使用 if (content == null) { content = message.getBody(); } return content; } ~~~ 结论: `Jackson2JsonMessageConverter`如果接收到的消息属性里面没有`content_type`属性,或者`content_type`值不包含json,则转换后的结果是`byte[]` ## Jackson2JsonMessageConverter详解续 上面我们提到的是将实体类型转换成Map或者List类型,这样转换没有多大意义,我们需要消费者将生产者的消息对象格式转换成对应的消息格式,而不是Map或者List对象,解决方案,看代码: 生成端代码: ~~~java /** * 生产者在发送json数据的时候,需要指定这个json是哪个对象,否则消费者收到消息之后,不知道要转换成哪个java对象 * * 指定方法: * 在消息header中,增加一个_TypeId_,value就是具体的java对象(全类名),一定是消费者所在系统的java对象全称 */ import com.fasterxml.jackson.databind.ObjectMapper; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.ComponentScan; import java.time.LocalDateTime; @ComponentScan public class Application { public static void sendOrder( RabbitTemplate rabbitTemplate) throws Exception{ Order order = new Order(); order.setId(1); order.setUserId(1000); order.setAmout(88d); order.setTime(LocalDateTime.now().toString()); ObjectMapper mapper = new ObjectMapper(); String json = mapper.writeValueAsString(order); System.out.println(json); MessageProperties messageProperties = new MessageProperties(); messageProperties.setContentType("application/json"); //指定的__TypeId__属性值必须是消费端的Order的全类名,如果不匹配则会报错。 messageProperties.getHeaders().put("__TypeId__","com.zhihao.miao.test.day10.Sender.Order"); Message message = new Message(json.getBytes(),messageProperties); rabbitTemplate.send("","zhihao.miao.order",message); } public static void main(String[] args) throws Exception{ AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(Application.class); RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class); System.out.println(rabbitTemplate); sendOrder(rabbitTemplate); context.close(); } } ~~~ 消费端的Handler改造: ~~~ import java.util.List; import java.util.Map; import com.zhihao.miao.test.day10.Sender.Order; public class MessageHandler { public void onMessage(byte[] message){ System.out.println("---------onMessage----byte-------------"); System.out.println(new String(message)); } public void onMessage(String message){ System.out.println("---------onMessage---String-------------"); System.out.println(message); } public void onMessage(Map order){ System.out.println("---------onMessage---map-------------"); System.out.println(order.toString()); } public void onMessage(Order order){ System.out.println("---------onMessage---Order-------------"); System.out.println(order); } public void onMessage(List orders){ System.out.println("---------onMessage---List-------------"); System.out.println(orders.toString()); } } ~~~ **总结**: * 生产者在发送json数据的时候,需要指定这个json是哪个对象,否则消费者收到消息之后,不知道要转换成哪个java对象 **指定方法**: * 在消息header中,增加一个*TypeId*,value就是具体的java对象(全类名),一定是消费者所在系统的java对象全称 ### 优化 我们发现生产者和消费者的耦合度太高,生产者需要知道消费者相应对应的全类名,如何去改造呢? 在消费端配置映射: ~~~java @Bean public SimpleMessageListenerContainer messageListenerContainer(ConnectionFactory connectionFactory){ SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); container.setConnectionFactory(connectionFactory); container.setQueueNames("zhihao.miao.order"); MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageHandler()); //指定Json转换器 Jackson2JsonMessageConverter jackson2JsonMessageConverter =new Jackson2JsonMessageConverter(); //消费端配置映射 Map<String, Class<?>> idClassMapping = new HashMap<>(); idClassMapping.put("order",Order.class); idClassMapping.put("user",User.class); DefaultJackson2JavaTypeMapper jackson2JavaTypeMapper = new DefaultJackson2JavaTypeMapper(); jackson2JavaTypeMapper.setIdClassMapping(idClassMapping); System.out.println("在jackson2JsonMessageConverter转换器中指定映射配置"); jackson2JsonMessageConverter.setJavaTypeMapper(jackson2JavaTypeMapper); adapter.setMessageConverter(jackson2JsonMessageConverter); //设置处理器的消费消息的默认方法 adapter.setDefaultListenerMethod("onMessage"); container.setMessageListener(adapter); return container; } ~~~ 消费者处理器Handler中增加入参数是User的方法: ~~~ import java.util.List; import java.util.Map; import com.zhihao.miao.test.day10.Sender.Order; import com.zhihao.miao.test.day10.Sender.User; public class MessageHandler { public void onMessage(byte[] message){ System.out.println("---------onMessage----byte-------------"); System.out.println(new String(message)); } public void onMessage(String message){ System.out.println("---------onMessage---String-------------"); System.out.println(message); } public void onMessage(Map order){ System.out.println("---------onMessage---map-------------"); System.out.println(order.toString()); } public void onMessage(Order order){ System.out.println("---------onMessage---Order-------------"); System.out.println(order); } public void onMessage(User user){ System.out.println("---------onMessage---user-------------"); System.out.println(user.toString()); } public void onMessage(List orders){ System.out.println("---------onMessage---List-------------"); System.out.println(orders.toString()); } } ~~~ 然后在生产端就可以指定对应的key,而不需要再去指定全类名了, ~~~java import com.fasterxml.jackson.databind.ObjectMapper; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.ComponentScan; import java.time.LocalDateTime; @ComponentScan public class Application { public static void sendOrder( RabbitTemplate rabbitTemplate) throws Exception{ Order order = new Order(); order.setId(1); order.setUserId(1000); order.setAmout(88d); order.setTime(LocalDateTime.now().toString()); System.out.println(order); ObjectMapper mapper = new ObjectMapper(); String json = mapper.writeValueAsString(order); System.out.println(json); MessageProperties messageProperties = new MessageProperties(); messageProperties.setContentType("application/json"); messageProperties.getHeaders().put("__TypeId__","order"); Message message = new Message(json.getBytes(),messageProperties); rabbitTemplate.send("","zhihao.miao.order",message); } public static void sendUser( RabbitTemplate rabbitTemplate) throws Exception{ User user = new User(); user.setUserId(1000); user.setAge(50); user.setUsername("zhihao.miao"); user.setPassword("123343"); System.out.println(user); ObjectMapper mapper = new ObjectMapper(); String json = mapper.writeValueAsString(user); System.out.println(json); MessageProperties messageProperties = new MessageProperties(); messageProperties.setContentType("application/json"); //指定消费端配置的key值就行了 messageProperties.getHeaders().put("__TypeId__","user"); Message message = new Message(json.getBytes(),messageProperties); rabbitTemplate.send("","zhihao.miao.order",message); } public static void main(String[] args) throws Exception{ AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(Application.class); RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class); System.out.println(rabbitTemplate); //sendOrder(rabbitTemplate); sendUser(rabbitTemplate); context.close(); } } ~~~ 进行测试发现结果符合我们预期。 **结论** 发送消息的时候,**TypeId**的值可以是java对象全称,也可以是映射的key 当消费者有配置映射key的时候,生产者既可以指定java对象全称,又可以是映射的key。如果消费者没有配置映射key,则只能指定java对象全称 ### Jackson2JsonMessageConverter详解续 如果消息类型是List或者Map类型的时候, 生产端: ~~~ public static void sendOrderList(RabbitTemplate rabbitTemplate) throws Exception{ Order order = new Order(); order.setId(1); order.setUserId(1000); order.setAmout(88d); order.setTime(LocalDateTime.now().toString()); Order order2 = new Order(); order2.setId(2); order2.setUserId(2000); order2.setAmout(99d); order2.setTime(LocalDateTime.now().toString()); List<Order> orderList = Arrays.asList(order,order2); ObjectMapper mapper = new ObjectMapper(); String json = mapper.writeValueAsString(orderList); MessageProperties messageProperties = new MessageProperties(); messageProperties.setContentType("application/json"); messageProperties.getHeaders().put("__TypeId__","java.util.List"); messageProperties.getHeaders().put("__ContentTypeId__","order"); Message message = new Message(json.getBytes(),messageProperties); rabbitTemplate.send("","zhihao.miao.order",message); } public static void sendOrderMap(RabbitTemplate rabbitTemplate) throws Exception{ Order order = new Order(); order.setId(1); order.setUserId(1000); order.setAmout(88d); order.setTime(LocalDateTime.now().toString()); Order order2 = new Order(); order2.setId(2); order2.setUserId(2000); order2.setAmout(99d); order2.setTime(LocalDateTime.now().toString()); Map<String,Object> orderMaps = new HashMap<>(); orderMaps.put("10",order); orderMaps.put("20",order2); ObjectMapper mapper = new ObjectMapper(); String json = mapper.writeValueAsString(orderMaps); MessageProperties messageProperties = new MessageProperties(); messageProperties.setContentType("application/json"); messageProperties.getHeaders().put("__TypeId__","java.util.Map"); messageProperties.getHeaders().put("__KeyTypeId__","java.lang.String"); messageProperties.getHeaders().put("__ContentTypeId__","order"); Message message = new Message(json.getBytes(),messageProperties); rabbitTemplate.send("","zhihao.miao.order",message); } ~~~ 消费端: ~~~ public void onMessage(List<Order> orders){ System.out.println("---------onMessage---List<Order>-------------"); orders.stream().forEach(order -> System.out.println(order)); } public void onMessage(Map<String,Object> orderMaps){ System.out.println("-------onMessage---Map<String,Object>------------"); orderMaps.keySet().forEach(key -> System.out.println(orderMaps.get(key))); } ~~~ **结论** 如果生产者发送的是list的json数据,则还需要增加一个`__ContentTypeId__`的header,用于指明List里面的具体对象。 如果生产者发送的是map的json数据,则需要指定`__KeyTypeId__`,`__ContentTypeId__`的header,用于指明map里面的key,value的具体对象。