Jackson2JsonMessageConverter · JAVA

[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的具體對象。