微服务框架 SpringCloud微服务架构 16 SpringAMQP 16.9 消息转换器

微服务框架

【SpringCloud+RabbitMQ+Docker+Redis+搜索+分布式，系统详解springcloud微服务技术栈课程|黑马程序员Java微服务】

SpringCloud微服务架构

16 SpringAMQP

16.9 消息转换器

16.9.1 一个案例

案例：测试发送Object类型消息

说明：在SpringAMQP的发送方法中，接收消息的类型是Object，也就是说我们可以发送任意对象类型的消息，SpringAMQP会帮我们序列化为字节后发送。

我们在consumer中利用@Bean声明一个队列：

@Bean
public Queue objectQueue(){
    return new Queue("object.queue");
}
1
2
3
4

现在直接重启 consumer 服务

OK，重启成功，直接查看RabbitMQ 控制台

可以看到 object.queue 队列已经生成了

在publisher中发送消息以测试：

@Test
public void testSendObjectQueue(){

    Map<String, Object> msg = new HashMap<>();
    msg.put("name","DingJiaxiong");
    msg.put("age",22);
    rabbitTemplate.convertAndSend("object.queue",msg);
}
1
2
3
4
5
6
7
8

直接运行

绿了， 消息发送成功 了

看看控制台

点进去查看详细信息

从内容类型可以看到， application/x-java-serialized-object 进行了java 序列化

【如何解决？】

16.9.2 消息转换器

Spring的对消息对象的处理是由org.springframework.amqp.support.converter.MessageConverter来处理的。而默认实现是SimpleMessageConverter，基于JDK的ObjectOutputStream完成序列化。

如果要修改只需要定义一个MessageConverter 类型的Bean即可。推荐用JSON方式序列化，步骤如下：

1. 在publisher服务引入依赖【其实干脆可以直接放到父工程】

<dependency>
    <groupId>com.fasterxml.jackson.coregroupId>
    <artifactId>jackson-databindartifactId>
dependency>
1
2
3
4

记得刷一下

2. 在publisher服务声明MessageConverter：

直接加 publisher 服务的启动类中加bean

package cn.itcast.mq;

import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
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 messageConverter(){

        return new Jackson2JsonMessageConverter();
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

OK

先到控制台把之前 那个消息清理掉

OK，这样消息就都没了

直接重新运行我们之前的测试

OK，发送成功，直接查看控制台

妙啊

【现在发送json 是完成了，接收json 该怎么做？】

在consumer服务引入Jackson依赖：

【依赖就不用导了，因为我们刚刚已经把依赖放到了父工程，所以每个儿子都有】

在consumer服务定义MessageConverter：

直接贴过来就行了，不用改

然后定义一个消费者，监听object.queue队列并消费消息：

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

OK，直接重启消费者服务

效果很明显，直接就把上次我们发那个消费掉了【这就是消息转换器】

16.9.3 总结

SpringAMQP中消息的序列化和反序列化是怎么实现的？

• 利用MessageConverter实现的，默认是JDK的序列化
• 注意发送方与接收方必须使用相同的MessageConverter