当前位置：首页 > article >正文

RabbitMQ配置消息转换器

article 2025/3/11 15:19:54

如果不配置转换器，在RabbitMQ中获取的消息乱码...

配置

1.引入依赖

<!--        消息队列JSON转换器-->
        <dependency>
            <groupId>com.fasterxml.jackson.dataformat</groupId>
            <artifactId>jackson-dataformat-xml</artifactId>
            <version>2.9.10</version>
        </dependency>

2.创建转换的Bean（这里的Jackson2JsonMessageConverter类的包不要引错）

import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class messageConverter {
    @Bean
    public MessageConverter messageConvertertest() {
        // 1.定义消息转换器
        Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
        // 2.配置自动创建消息id，用于识别不同消息，也可以在业务中基于ID判断是否是重复消息
        jackson2JsonMessageConverter.setCreateMessageIds(true);
        return jackson2JsonMessageConverter;
    }
}

测试

写入生产者和消费者

生产者我是在测试中写入

    @Test
    public void testSendMapMessage()   {
        Map<String,Object> msg = new HashMap<>();
        msg.put("id", 1);
        msg.put("name", "柳岩");
        msg.put("age", 21);
        rabbitTemplate.convertAndSend("test.exchange", "test.key", msg);
    }

这里指定的交换机是test.exchange，绑定的routingkey是test.key,通过routingkey可以找到对应的消息队列

消费者

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "test.json.queue", durable = "true"),
            exchange = @Exchange(name = "test.exchange", type = ExchangeTypes.TOPIC),
            key = "test.key"
    ))
    public void ToExchangeToQueue3(Map<String, Object> message){
        System.out.println("消费者接收到test.json.queue消息：【" + message + "】");
    }

消费者的交换机名字需要与生产者交换机名字一致，key需与生产者的routingkey一致，交换机类型设置了topic类型，通过交换机名字和绑定的routingkey可以转发到队列test.json.queue,然后将消息输出

测试发送消息

另外可以再创建一个消费者，绑定test.json.queue队列

    @RabbitListener(queues = "test.json.queue")
    public void listenSimpleQueueMessage(Map<String, Object> msg) throws InterruptedException {
        System.out.println("消费者2：【" + msg + "】");
    }

然后模拟发送十条消息


    @Test
    public void testSendMap()   {
        Map<String,Object> msg = new HashMap<>();
        //模拟发送十条消息
        int count = 10;
        for (int i = 0; i < count; i++) {
            msg.put("id", i);
            msg.put("name", "柳岩");
            msg.put("age", 21);
            rabbitTemplate.convertAndSend("test.exchange", "test.key", msg);
        }
    }

输出结果