快捷導(dǎo)航

一文帶你了解RabbitMQ消息轉(zhuǎn)換器

更新時(shí)間：2023年04月02日 09:20:00 作者：小威要向諸佬學(xué)習(xí)呀

這篇文章主要為大家詳細(xì)介紹了RabbitMQ中消息轉(zhuǎn)換器的相關(guān)知識(shí)，文中的示例代碼講解詳細(xì)，具有一定的借鑒價(jià)值，感興趣的小伙伴可以了解一下

在SpringAMQP的發(fā)送方法中，接收消息的類(lèi)型是Object，也就是說(shuō)我們可以發(fā)送任意對(duì)象類(lèi)型的消息，SpringAMQP會(huì)幫我們序列化為字節(jié)后發(fā)送。

我們以實(shí)際例子來(lái)做一個(gè)演示，首先在項(xiàng)目中的FanoutConfig配置類(lèi)聲明一個(gè)新的隊(duì)列：

package cn.itcast.mq.config;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class FanoutConfig {
    @Bean
    public Queue objectQueue(){
        return new Queue("object.queue");
    }
}

接著我們?cè)谏a(chǎn)者模塊中編寫(xiě)發(fā)送消息的測(cè)試代碼：

package cn.itcast.mq.spring;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
import java.util.HashMap;
import java.util.Map;

@SpringBootTest
@RunWith(SpringRunner.class)
public class SpringAmqpTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    public void testSendObjectQueue(){
        Map<String,Object> msg =new HashMap<>();//創(chuàng)建一個(gè)map集合
        msg.put("name","小威");
        msg.put("age",21);
        rabbitTemplate.convertAndSend("object.queue",msg);//發(fā)送消息
    }
}

消息發(fā)送代碼編寫(xiě)完成后，重啟項(xiàng)目，打開(kāi)RabbitMQ的虛擬機(jī)對(duì)應(yīng)地址可以看到我們編寫(xiě)的name和age出現(xiàn)了特殊的情況：

這是因?yàn)镽abbitMQ只支持字節(jié)的序列化方式，spring支持發(fā)送object對(duì)象消息，由上圖可以看到content_type的轉(zhuǎn)化方式，spring將封裝的map對(duì)象序列化了一長(zhǎng)串字符串。

那么我們?cè)撊绾谓鉀Q這個(gè)問(wèn)題呢？

首先，Spring的消息對(duì)象的處理是由org.springframework.amqp.support.converter.MessageConverter來(lái)處理的。默認(rèn)實(shí)現(xiàn)是SimpleMessageConverter，基于JDK的ObjectOutputStream完成序列化。

因此，我們需要定義一個(gè)MessageConverter 類(lèi)型的Bean，可以使用json的方式序列化，在父工程中引入json的依賴(lài)：

        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
        </dependency>

然后在生產(chǎn)者模塊的啟動(dòng)類(lèi)中聲明bean：

package cn.itcast.mq;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.Jackson2XmlMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
@SpringBootApplication
public class PublisherApplication {
    public static void main(String[] args) {
        SpringApplication.run(PublisherApplication.class);
    }
    @Bean
    public MessageConverter jsonMessageConverter(){
        return new Jackson2JsonMessageConverter();
    }
}

再次重啟生產(chǎn)者模塊可以看到我們想要的消息對(duì)象結(jié)果了：

對(duì)于消費(fèi)者也同理，首選引入依賴(lài)，接著在啟動(dòng)類(lèi)中編寫(xiě)和生產(chǎn)者一樣的代碼，最后定義一個(gè)消費(fèi)者監(jiān)聽(tīng)該隊(duì)列：

    @RabbitListener(queues = "object.queue")
    public void listenObjectQueue(Map<String,Object> msg){
        System.out.println("接受到object.queue的消息："+msg);
    }

啟動(dòng)消費(fèi)者模塊可以看到控制臺(tái)輸出的結(jié)果：