rabbitmq-spring-boot-start版本優化升級

文章目錄

1.前言
2.優化升級內容
3.依賴
4.使用
- 4.1發送消息代碼示例
- 4.2消費監聽代碼示例
- 4.3 brock中的消息
5.RabbmitMq的MessageConverter消息轉換器
- 5.1默認行為
- 5.2JDK 序列化的缺點
- 5.3使用 JSON 進行序列化
6.總結

1.前言

????由于之前手寫了一個好用的rabbitmq-spring-boot-start啟動器，雖然實現了相關的功能，但是還有一點小問題，這個也是最近在項目中使用我寫的這個組件發現的一個問題，所以修復升級了，maven依賴已經退推送到中央倉庫上了，只需要引入依賴，然后簡單的配置就可以實現發消息了，使用上會有一點需要注意的，請聽下文分解。

2.優化升級內容

????升級了RabbitService類中相關發送MQ消息方法上的String msg的類型改為了Object msg，之前在項目中使用的時候發送消息的時候這個參數是String類型會有啥問題，如果要發送一個對象，將這個對象序列化成JSON字符串，最終消費者監聽的地方反序列化解析不來，因為收到過的是一個String不是一個JSON的字符串，這個字符串中會含有轉義\，還有就是字符串的前后會有一個引號（"）,導致使用FastJson反序列化解析失敗，下面是發送消息的demo代碼

 Student student = new Student();student.setName("張三");student.setSex("男");student.setAge(18);RabbitTemplate rabbitTemplate = (RabbitTemplate) ZlfMqSpringUtils.getBean(ZlfMqRegistrarBeanNamePrefix.rabbitTemplatePrefix + 0);rabbitService.sendDelayed(rabbitTemplate, "zlf.delay.test1", "delay.test1.key", JSON.toJSONString(student), 60);

????消費者監聽的地方解析代碼：

    @RabbitHandler@RabbitListener(queues = "zlf.delay.test1", containerFactory = ZlfMqRegistrarBeanNamePrefix.simpleRabbitListenerContainerFactory + 0)public void mqConsumer(Message message, Channel channel) throws IOException {String msg = new String(message.getBody(), "UTF-8");try {log.info("mqConsumer=====>msg1:{}", msg);msg = StringEscapeUtils.unescapeJava(msg);log.info("mqConsumer=====>msg2:{}", msg);msg = msg.substring(1, msg.length() - 1);log.info("mqConsumer=====>msg3:{}", msg);if (StringUtils.isNotEmpty(msg)) {Student student = JSON.parseObject(message.getBody(), Student.class);log.info("mqConsumer=====>studen:{}", JSON.toJSONString(student));//TODO 業務處理}} catch (Exception e) {log.error("mqConsumer消費異常:{}", e.getMessage());} finally {channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);}//channel.basicNack(); 不ack//channel.basicReject(); 拒絕}

????這種方法雖然可以解決這個消費者反序列化的問題，但是只是對于發送FastJson序列化的對象中的字段不含有特殊的轉義字符，如果有會被StringEscapeUtils.unescapeJava方法去除轉義，我這種處理是業務上發送那個FastJson序列化的對象字段沒有特殊的轉義字符，所以這種搞是可以的，于是乎我就抽了一點時間優化升級了下，然后也寫了個demo測試了一下，非常的耐思。

3.依賴

????下面兩個依賴任選其一引入就可以使用了：

<dependency><groupId>io.gitee.bigbigfeifei</groupId><artifactId>rabbitmq-spring-boot-start</artifactId><version>1.1</version>
</dependency>
或者
<dependency><groupId>io.github.bigbigfeifei</groupId><artifactId>rabbitmq-spring-boot-start</artifactId><version>1.1</version>
</dependency>

4.使用

4.1發送消息代碼示例

????只需要發送一個對象就可以了

 Student student = new Student();student.setName("張三");student.setSex("男");student.setAge(18);RabbitTemplate rabbitTemplate = (RabbitTemplate) ZlfMqSpringUtils.getBean(ZlfMqRegistrarBeanNamePrefix.rabbitTemplatePrefix + 0);rabbitService.sendDelayed(rabbitTemplate, "zlf.delay.test1", "delay.test1.key", student, 60);

4.2消費監聽代碼示例

/*** 延遲插件實現延遲隊列監聽隊列消息** @param message* @param channel* @throws IOException*/@RabbitHandler@RabbitListener(queues = "delay.test1", containerFactory = ZlfMqRegistrarBeanNamePrefix.simpleRabbitListenerContainerFactory + 0)public void mqConsumer1(Message message, Channel channel) throws IOException {try {Student student = JSON.parseObject(message.getBody(), Student.class);log.info("mqConsumer1=====>student:{}", JSON.toJSONString(student));//TODO 業務處理} catch (Exception e) {log.error("mqConsumer1消費異常:{}", e.getMessage());} finally {channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);}//channel.basicNack(); 不ack//channel.basicReject(); 拒絕}

4.3 brock中的消息

????本個start使用的是Jackson2JsonMessageConverter消息轉換器來序列化發送消息實體為標準的JSON數據，源碼在ZlfRabbitMqRegistrar類中registerBeanDefinitions方法中，代碼如下：

 //構建發送的RabbitTemplate實例關聯連接工廠Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();RabbitTemplate rabbitTemplate = new RabbitTemplate(cachingConnectionFactory);rabbitTemplate.setMessageConverter(jackson2JsonMessageConverter);

????為每一個對應的rabbitTemplate對象Bean設置了一個Jackson2JsonMessageConverter消息轉換器，這樣就可以支持JSON數據的序列化和反序列化了。

5.RabbmitMq的MessageConverter消息轉換器

????RabbmitMq默認實現是 SimpleMessageConverter，該類基于 JDK 的 ObjectOutputStream 完成序列化。

5.1默認行為

發送消息時：消息會通過 SimpleMessageConverter 轉換為字節流。
接收消息時：消息會被自動反序列化回Java對象。
默認的序列化方式：Spring AMQP默認使用JDK自帶的序列化機制。消息格式為 application/x-java-serialized-object。

5.2JDK 序列化的缺點

????默認的 JDK 序列化雖然可以轉換任意對象，但它存在幾個顯著的問題：

????安全漏洞：JDK序列化存在反序列化攻擊的風險。
????體積過大：序列化后的字節流體積較大，占用帶寬和存儲空間。
????可讀性差：序列化后的字節流不可讀，無法直觀查看消息內容。

5.3使用 JSON 進行序列化

????為了避免JDK序列化的缺點，可以使用JSON作為消息的序列化格式，這樣：

????數據體積較小。
????數據可讀性好。
????提高了消息的安全性。

????所以默認的 JDK 的 ObjectOutputStream 完成序列化時候該如何處理：

    //  字節碼轉化為對象public Object getObjectFromBytes(byte[] objBytes) throws Exception {if (objBytes == null || objBytes.length == 0) {return null;}ByteArrayInputStream bi = new ByteArrayInputStream(objBytes);ObjectInputStream oi = new ObjectInputStream(bi);return oi.readObject();}

????發送消息時：

Student student = new Student();student.setName("張三");student.setSex("男");student.setAge(18);
byte[] e = getBytesFromObject(student);
Message message = MessageBuilder.withBody(e).setContentType(MessageProperties.CONTENT_TYPE_JSON).build();
//或者直接發一個對象，然后消費監聽的時候從byte[]中讀取反序列化為一個java對象

????消費者監聽調用getObjectFromBytes方法反序列化為一個java對象

 Student student = (Student)getObjectFromBytes(message.getBody());

????這種原始的方式不推薦使用，還是推薦使用JSON的方式。

????MessageConverter接口還有很多的實現類，可以根據你自己的需求去使用或者是自定義一個消息轉換器，這個難度也比較的大，所以也不推薦去搞。

????SimpleRabbitListenerContainerFactory、DirectRabbitListenerContainerFactory、RabbitTemplate這三個類中都有setMessageConverter的方法，只要從容器中獲取對應的Bean，讓然把消息轉換器的對象設置進去就可以了，切入點可以使用一個類實現implements InitializingBean接口，在afterPropertiesSet中取出對應的bean來設置

public void afterPropertiesSet() throws Exception {//設置SimpleRabbitListenerContainerFactory的消息轉換器為Jackson2JsonMessageConverter支持json格式的數據轉換SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory = (SimpleRabbitListenerContainerFactory) ZlfMqSpringUtils.getBean(ZlfMqRegistrarBeanNamePrefix.simpleRabbitListenerContainerFactory + 0);simpleRabbitListenerContainerFactory.setMessageConverter(new Jackson2JsonMessageConverter());}

????或者：

RabbitTemplate rabbitTemplate = (RabbitTemplate) ZlfMqSpringUtils.getBean(ZlfMqRegistrarBeanNamePrefix.rabbitTemplatePrefix + 0);rabbitService.sendDelayed(rabbitTemplate, "zlf.delay.test1", "delay.test1.key", JSON.toJSONString(student), 60);