3.6.基于注解声明队列和交换机

简介: 通过SpringAMQP,程序可自动声明队列和交换机,避免手动配置出错。支持@Bean方式和更简洁的@RabbitListener注解方式,实现启动时自动创建。同时,默认JDK序列化存在体积大、不安全等问题,需优化消息转换器。

在之前我们都是基于RabbitMQ控制台来创建队列、交换机。但是在实际开发时,队列和交换机是程序员定义的,将来项目上线,又要交给运维去创建。那么程序员就需要把程序中运行的所有队列和交换机都写下来,交给运维。在这个过程中是很容易出现错误的。
因此推荐的做法是由程序启动时检查队列和交换机是否存在,如果不存在自动创建。
3.8.1. 基本API的方式(了解)
3.8.1.1 介绍
SpringAMQP提供了一个Queue类,用来创建队列:

SpringAMQP还提供了一个Exchange接口,来表示所有不同类型的交换机:

我们可以自己创建队列和交换机,不过SpringAMQP还提供了ExchangeBuilder来简化这个过程:

而在绑定队列和交换机时,则需要使用BindingBuilder来创建Binding对象:

3.8.1.2 API方式声明举例
下边是fanout示例
在consumer中创建一个类,声明队列和交换机:
package com.itheima.consumer.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 {
/**

 * 声明交换机
 * @return Fanout类型交换机
 */
@Bean
public FanoutExchange fanoutExchange(){
    return new FanoutExchange("hmall.fanout");
}

/**
 * 第1个队列
 */
@Bean
public Queue fanoutQueue1(){
    return new Queue("fanout.queue1");
}

/**
 * 绑定队列和交换机
 */
@Bean
public Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){
    return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
}

/**
 * 第2个队列
 */
@Bean
public Queue fanoutQueue2(){
    return new Queue("fanout.queue2");
}

/**
 * 绑定队列和交换机
 */
@Bean
public Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){
    return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
}

}
下边是direct示例
direct模式由于要绑定多个KEY,会非常麻烦,每一个Key都要编写一个binding:
package com.itheima.consumer.config;

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class DirectConfig {

/**
 * 声明交换机
 * @return Direct类型交换机
 */
@Bean
public DirectExchange directExchange(){
    return ExchangeBuilder.directExchange("hmall.direct").build();
}

/**
 * 第1个队列
 */
@Bean
public Queue directQueue1(){
    return new Queue("direct.queue1");
}

/**
 * 绑定队列和交换机
 */
@Bean
public Binding bindingQueue1WithRed(Queue directQueue1, DirectExchange directExchange){
    return BindingBuilder.bind(directQueue1).to(directExchange).with("red");
}
/**
 * 绑定队列和交换机
 */
@Bean
public Binding bindingQueue1WithBlue(Queue directQueue1, DirectExchange directExchange){
    return BindingBuilder.bind(directQueue1).to(directExchange).with("blue");
}

/**
 * 第2个队列
 */
@Bean
public Queue directQueue2(){
    return new Queue("direct.queue2");
}

/**
 * 绑定队列和交换机
 */
@Bean
public Binding bindingQueue2WithRed(Queue directQueue2, DirectExchange directExchange){
    return BindingBuilder.bind(directQueue2).to(directExchange).with("red");
}
/**
 * 绑定队列和交换机
 */
@Bean
public Binding bindingQueue2WithYellow(Queue directQueue2, DirectExchange directExchange){
    return BindingBuilder.bind(directQueue2).to(directExchange).with("yellow");
}

}
3.8.2. 基于注解声明
基于@Bean的方式声明队列和交换机比较麻烦,Spring还提供了基于注解方式来声明。
例如,我们同样声明Direct模式的交换机和队列:
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "direct.queue1"),
exchange = @Exchange(name = "hmall.direct", type = ExchangeTypes.DIRECT),
key = {"red", "blue"}
))
public void listenDirectQueue1(String msg){
System.out.println("消费者1接收到direct.queue1的消息:【" + msg + "】");
}

@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "direct.queue2"),
exchange = @Exchange(name = "hmall.direct", type = ExchangeTypes.DIRECT),
key = {"red", "yellow"}
))
public void listenDirectQueue2(String msg){
System.out.println("消费者2接收到direct.queue2的消息:【" + msg + "】");
}
测试方法非常简单,把hmall.direct交换机及绑定的队列通过rabbitmq控制台删除,重启consumer服务后hmall.direct交换机及绑定的队列会自动创建。
是不是简单多了。
再试试Topic模式:
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "topic.queue1"),
exchange = @Exchange(name = "hmall.topic", type = ExchangeTypes.TOPIC),
key = "china.#"
))
public void listenTopicQueue1(String msg){
System.out.println("消费者1接收到topic.queue1的消息:【" + msg + "】");
}

@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "topic.queue2"),
exchange = @Exchange(name = "hmall.topic", type = ExchangeTypes.TOPIC),
key = "#.news"
))
public void listenTopicQueue2(String msg){
System.out.println("消费者2接收到topic.queue2的消息:【" + msg + "】");
}
3.7.消息转换器
3.9.1 测试默认转换器
Spring的消息发送代码接收的消息体是一个Object:

而在数据传输时,它会把你发送的消息序列化为字节发送给MQ,接收消息的时候,还会把字节反序列化为Java对象。
只不过,默认情况下Spring采用的序列化方式是JDK序列化。众所周知,JDK序列化存在下列问题:
● 数据体积过大
● 有安全漏洞
● 可读性差
我们来测试一下。
1)创建测试队列:object.queue

2)发送消息
我们在publisher模块的SpringAmqpTest中新增一个消息发送的代码,发送一个Map对象:
@Test
public void testSendMap() throws InterruptedException {
// 准备消息
Map msg = new HashMap<>();
msg.put("name", "柳岩");
msg.put("age", 21);
// 发送消息
rabbitTemplate.convertAndSend("object.queue", msg);
}
注意,这里我们先不要给这个队列添加消费者,我们要查看消息体的格式。
发送消息后查看控制台:

可以看到消息格式非常不友好。

相关文章
|
Java Maven
启动SpringBoot 错误: 找不到或无法加载主类 com.xx.xxApplication
1、在项目名称上右击–Reload from disk 2、清理项目里面的class文件进行重新编译, 鼠标点击右侧Maven,找到项目名称,然后点击Lifecycle中的clean/compile/install
3391 1
|
9月前
|
消息中间件 JSON Java
3.9.2 配置JSON转换器
为优化消息序列化,采用JSON格式替代JDK默认方式。在publisher和consumer中引入Jackson依赖(若已引入web starter则无需重复添加),并配置`Jackson2JsonMessageConverter`作为消息转换器,启用自动生成消息ID。通过MQ控制台验证消息结构,并在消费者端使用Map接收发送的Object消息,确保收发一致。
|
10月前
|
消息中间件 网络安全 数据安全/隐私保护
我用 Docker 部署 RabbitMQ 踩了 3 个大坑,10 分钟搞定的记录
部署RabbitMQ踩坑记:从Docker安装、镜像拉取到容器启动,亲历5大常见陷阱。分享一键脚本、推荐镜像标签、关键参数配置及Cookie权限修复方案,结合文档避坑指南,助你10分钟快速稳定部署。
1568 6
|
机器学习/深度学习 人工智能 弹性计算
阿里云《AI 剧本生成与动画创作》解决方案深度评测
阿里云《AI 剧本生成与动画创作》解决方案深度评测
760 7
|
Java 数据库连接 数据库
【潜意识Java】深度分析黑马项目《苍穹外卖》在Java学习中的重要性
《苍穹外卖》项目对Java学习至关重要。它涵盖了用户管理、商品查询、订单处理等模块,涉及Spring Boot、MyBatis、Redis等技术栈。
2262 4
|
网络协议 安全 API
家用路由器DNS被恶意篡改?教你如何应对
近期大量用户反馈家用路由器出现解析失效,怀疑部分家用路由器的DNS配置遭遇了非正常变动。我们建议家用路由器用户将本地DNS改成更安全可靠的服务器,如阿里云公共DNS;企业APP终端用户使用公共DNS企业版服务。
|
NoSQL Java Redis
RedisTemplate 实现基于 Value 操作的简易锁机制
RedisTemplate 实现基于 Value 操作的简易锁机制
1392 0
|
存储 安全 Java
Hashtable 和 HashMap 的区别
【8月更文挑战第22天】
1559 0
|
Web App开发 开发者
Apifox 网页端调试本地接口
Apifox 网页端调试本地接口
2852 0
|
缓存 Linux 开发工具
centos设置ntp服务同步目标服务器时间
【7 月更文挑战第 1天】linux+centos设置ntp服务同步目标服务器时间