目录
一.准备工作
本次入门是在原先创建的SpringBoot基础上添加可以点击此处查看
1.项目结构分析
pom.xml
添加RabbitMq依赖文件
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
<!-- <version>2.6.1</version>--><!--由于前面已经定义了springboot的版本号这里省略-->
</dependency>
由于前面已经定义了springboot的版本号这里省略
项目中还使用到了hutool工具:一个Java基础工具类,对文件、流、加密解密、转码、正则、线程、XML等JDK方法进行封装,组成各种Util工具类
<dependency>
<groupId>cn.hutool</groupId>
<artifactId>hutool-all</artifactId>
<version>5.7.17</version>
</dependency>
配置application.yml
rabbitmq: #设置rabbitMq的配置
host: 127.0.0.1
port: 5672
username: guest
password: guest
virtual-host: /
2.了解相关知识
Broker:简单来说就是消息队列服务器实体。 Exchange:消息交换机,它指定消息按什么规则,路由到哪个队列。
Queue:消息队列载体,每个消息都会被投入到一个或多个队列。
Binding:绑定,它的作用就是把exchange和queue按照路由规则绑定起来。
Routing Key:路由关键字,exchange根据这个关键字进行消息投递。
vhost:虚拟主机,一个broker里可以开设多个vhost,用作不同用户的权限分离。 producer:消息生产者,就是投递消息的程序。
consumer:消息消费者,就是接受消息的程序。
channel:消息通道,在客户端的每个连接里,可建立多个channel,每个channel代表一个会话任务。
二.Direct模式
Direct模式相当于点对点模式,一个消息被发送者发送后,会被转发到通过routingkey绑定到具体消息队列中,然后被一个接收者接收!
1.Direct配置编写
在config创建DirectConfig类用于配置队列和路由等信息绑定
@Configuration
public class DirectConfig {
/**
* 路由关键字
*/
public static final String ROUTING_ORDER="direct_order_routing";//订单的路由
/**
* 创建队列
*/
public static final String QUEUE_OREDER="direct_order_message";//订单的队列
/**
* 配置交换机
*/
public static final String DIRECT_EXCHANGE="direct";//direct类型交换机
/**
* 声明队列
* @return
*/
@Bean
public Queue directQueue(){
//持久 - 如果我们声明一个持久队列,则为真(该队列将在服务器重启后继续存在)
return new Queue(QUEUE_OREDER,false);
}
/**
* 声明交换机
* @return
*/
@Bean
public DirectExchange directExchange(){
//交换器名称、是否持久化、是否自动删除
return new DirectExchange(DIRECT_EXCHANGE,true,false);
}
/***
* 队列绑定到交换机上
* 并且设置路由关键字
*/
@Bean
public Binding binding(Queue queue,DirectExchange exchange){
return BindingBuilder.bind(queue).to(exchange).with(ROUTING_ORDER);
}
}
2.创建消费者
Consumer包下创建订单的消费者OrderConsumer
@RabbitListener
可以作用在类上面,需配合 @RabbitHandler 注解一起使用也可以配置方法上就不需要@RabbitHandler
@RabbitListener(queues = {"direct_order_message"})
可以监听多个队列我这里只监听了一个名称是direct_order_message
的队列
@Component
@RabbitListener(queues = {"direct_order_message"})//队列名称(direct_order_message)
public class OrderConsumer {
private static final Logger logger= LoggerFactory.getLogger(OrderConsumer.class);
@RabbitHandler//区分消息的内容信息运行不同的方法
public void handler(Map message){
//logger.info("OrderConsumer处理订单信息:"+message.toString());
System.out.println("OrderConsumer处理订单信息:"+message.toString());
}
}
3.创建生产者
Producer包下创建生产者用于生产信息SenderOrderProducer
/**
* 生产者--发送消息
*/
@Component
public class SenderOrderProducer {
private static final Logger logger = LoggerFactory.getLogger(SenderOrderProducer.class);
@Autowired
private RabbitTemplate rabbitmqTemplate;//创建rqbbit模板对象
/**
* 发送信息
* @param exchange 交换机名称
* @param routingKey 路由键
* @param message 数据信息
*/
public void send(String exchange, String routingKey,Object message){
//logger.info("存入队列的消息:"+message);
System.out.println("存入队列的消息:"+message);
rabbitmqTemplate.convertAndSend(exchange,routingKey,message);
}
}
4.创建Direct接口测试
Controller包下创建RabbitMqController用于对外调用查看信息使用到swagger注解生成在线文档(可以看往前文章)
@Api(description = "RabbitMq")
@Component
@RequestMapping("rabbitMq")
public class RabbitMqController {
@Autowired
private SenderOrderProducer orderProducer;//生产者对象
@ApiOperation(value = "订单信息")
@PostMapping("transferOrder")
public ResponseEntity<Void> testOrder(@ApiParam(value = "数据信息",required = true) @RequestParam String message){
Map map=new HashMap();
map.put("UUID", UUID.randomUUID());
map.put("MESSAGE",message);
map.put("CREATEDATE", DateTime.now());//2021-12-21 10:35:31
//交换机名称 路由关键字 传递的数据
orderProducer.send(DirectConfig.DIRECT_EXCHANGE,DirectConfig.ROUTING_ORDER,map);
return ResponseEntity.noContent().build();
}
}
运行服务服务swagger网页:ip:服务的端口/swagger-ui.html
发送请求
队列中的信息被消费
5.模拟多个消费者
模拟下多个消费者来获取rabbit的信息
我们把OrderConsumer类复制一份,修改打印的信息做一个标识和之前的消费者做个区分
由于请求只会往队列插入一条信息,我们需要改造RabbitMqController
循环插入信息
@ApiOperation(value = "订单信息")
@PostMapping("transferOrder")
public ResponseEntity<Void> testOrder(@ApiParam(value = "数据信息",required = true) @RequestParam String message){
for (int i = 0; i < 6; i++) {
Map map=new HashMap();
map.put("UUID", UUID.randomUUID());
map.put("MESSAGE",message);
map.put("CREATEDATE", DateTime.now());//2021-12-21 10:35:31
//交换机名称 路由关键字 传递的数据
orderProducer.send(DirectConfig.DIRECT_EXCHANGE,DirectConfig.ROUTING_ORDER,map);
}
return ResponseEntity.noContent().build();
}
调用swagger运行接口,可以看出往队列添加了6条数据,并且消费者2个已经分别对信息进行消费,并且他们不会消费重复数据可以看uuid是否有重复来区分是否重复消费
创建一个对象用于接受前端传递过来的json数据,创建Pojo包下面在创建一个订单实体类
注意:一定要把实体类实现Serializable接口,不然传递给rabbit会报错
/**
* 订单实体类
*/
public class OrderEntity implements Serializable {
/**
* 订单号
*/
private String orderNo;
/**
* 创建时间
*/
private String createTime;
/**
* 金额
*/
private BigDecimal amount;
//省略get/set/toString方法
}
RabbitMqController类添加新接口
这一次我们传递的信息是订单对象,不像之前传递过去的是字符串
@ApiOperation(value = "订单信息")
@PostMapping("createOrder")
public ResponseEntity<Void> createOrder(@ApiParam(value = "订单数据",required = true) @RequestBody OrderEntity orderEntity){
for (int i = 0; i < 6; i++) {
Map map=new HashMap();
map.put("UUID", UUID.randomUUID());
map.put("MESSAGE",orderEntity);//传递订单对象
map.put("CREATEDATE", DateTime.now());//2021-12-21 10:35:31
//交换机名称 路由关键字 传递的数据
orderProducer.send(DirectConfig.DIRECT_EXCHANGE,DirectConfig.ROUTING_ORDER,map);
}
return ResponseEntity.noContent().build();
}
重新运行项目,访问swagger
数据被消费者分别消费成功
测试没有消费者,只把数据传递到队列中
把2个消费者的@RabbitListener
和@RabbitHandler
都注释
重启项目在运行接口把信息传递到队列中,因为消费者已经没有了所以数据会存放在rabbit中
我们可以访问Rabbiymq的地址查看:运行rabbitmq的ip:15672
没有改动的话一般都是端口15672 默认用户名和密码都是:guest
我们在调用接口
看控制台又插入6条数据到队列中
在通过浏览器查看rabbitmq管理界面队列已经有12条数据了
现在我们重启没有调用接口是不会往队列在创建数据,这个时候我们把消费者的注释都去除,相当于有2个消费者会消费队列里面的信息数据
消费者1:
消费者2:
运行后:效果是消费者1消费了所有数据出现这种情况也可能是因为消费者1先被spring创建了导致消费者2被创建出来后,可能已经全部被先创建的消费者1消费光了(这个的可能性大些)
我先排除下是不是因为消费者2没被创建出来所以导致消费者1全部消费完成.
在我只要请求一下接口看一下会不会有消费者2出来消费数据
可以看出消费者2是存在的,并且这6条数据是for循环调用插入队列中,可以看出for在运行第4条还没有插入完成,消费端就已经开始消费信息了
6.单消费者运行耗时程序
只用一个消费者消费信息,我们通过休眠模拟业务这里休眠9秒
我们现在调用接口插入6条数据到队列中
可以看出由于业务比较耗时已经在一条一条的处理中
rabbitmq管理界面也展示出正在消费中
三.Topic转发模式
可以看出消息是通过routing key与binding key的匹配关系进行路由的匹配,key可以由多个单词进行匹配通过.
来隔开,其中需要注意的是2个通配符
:#
匹配一个或者多个,*
匹配一个
1.前期准备
注意:为了测试Topic转发模式我们需要把Direct模式用到的类都删除,避免影响到Topic的测试,我这的做法是把Direct使用到的类@Configuration
和@Component
注释这样spring就不会加载这个对象
DirectConfig配置类注释上
OrderConsumer消费者类注释上
把订单的消费者2删除掉
在把RabbitMqController的配置注释上这样swagger就不会展示Direct测试接口
2.Topic配置编写
在config包下创建TopicConfig配置,创建路由关键字,队列名称,创建交换机,在配置交换机和队列的关系.
注意:在Spring中通过@Bean
创建对象时,如果多个bean的类型是一样的我们需要使用@Qualifier()
来指定使用的是那个@Bean,如果@Bean(name="cs")
没有指定name属性那么@Qualifier()里面填写的就是方法的名称,你可以在这个绑定交换机和队列的方法上看到参数有使用这个注解@Qualifier()
/**
* Topic模式
*/
@Configuration
public class TopicConfig {
/**
* 路由关键字
*/
public static final String ROUTING_ORDER="direct.order.routing";//订单的路由关键字
public static final String ROUTING_WMS="direct.wms.routing";//库存的路由关键字
/**
* 创建队列名称
*/
public static final String QUEUE_TPOIC_ORDER="topic_order_message";//订单的队列
public static final String QUEUE_TPOIC_WMS="topic_wms_message";//库存的队列
/**
* 配置交换机
*/
public static final String TOPIC_EXCHANGE="topic";//topic类型交换机
/**
* 声明队列
* @return
*/
@Bean
public Queue topicQueueOrder(){
//true持久 - 如果我们声明一个持久队列,则为真(该队列将在服务器重启后继续存在)
return new Queue(QUEUE_TPOIC_ORDER,true);
}
@Bean
public Queue topicQueueWms(){
//true持久 - 如果我们声明一个持久队列,则为真(该队列将在服务器重启后继续存在)
return new Queue(QUEUE_TPOIC_WMS,true);
}
/**
* 声明交换机
* @return
*/
@Bean
public TopicExchange topicExchange(){
//交换器名称、是否持久化、是否自动删除
return new TopicExchange(TOPIC_EXCHANGE,true,false);
}
/***
* 队列绑定到交换机上
* 并且设置路由关键字
*/
@Bean
public Binding bindingExchangeOrder(@Qualifier("topicQueueWms") Queue queue, TopicExchange topicExchange){
return BindingBuilder.bind(queue).to(topicExchange).with(ROUTING_WMS);
}
/***
* 队列绑定到交换机上
* 并且设置路由关键字
*/
@Bean
public Binding bindingExchangeWms(@Qualifier("topicQueueOrder") Queue queue, TopicExchange topicExchange){
return BindingBuilder.bind(queue).to(topicExchange).with(ROUTING_ORDER);
}
}
3.创建Topic接口测试
RabbitMqController创建2个接口一个是发一条数据到队列,一个是多条数据到队列并且对象是订单对象
生产者类不需要改动,我们只要把传人的数据改成TopicConfig的交换机和路由的键
@ApiOperation(value = "Topic:订单信息")
@PostMapping("transferOrderTopic")
public ResponseEntity<Void> transferOrderTopic(@ApiParam(value = "数据信息",required = true) @RequestParam String message){
Map map=new HashMap();
map.put("UUID", UUID.randomUUID());
map.put("MESSAGE",message);
map.put("CREATEDATE", DateTime.now());//2021-12-21 10:35:31
//交换机名称 路由关键字 传递的数据
orderProducer.send(TopicConfig.TOPIC_EXCHANGE,TopicConfig.ROUTING_ORDER,map);
return ResponseEntity.noContent().build();
}
@ApiOperation(value = "Topic:仓库信息创建多条到队列中")
@PostMapping("createWmsTopic")
public ResponseEntity<Void> createWmsTopic(@ApiParam(value = "仓库数据",required = true) @RequestBody OrderEntity wmsEntity){
for (int i = 0; i < 6; i++) {
Map map=new HashMap();
map.put("UUID", UUID.randomUUID());
map.put("MESSAGE",wmsEntity);
map.put("CREATEDATE", DateTime.now());//2021-12-21 10:35:31
//交换机名称 路由关键字 传递的数据
orderProducer.send(TopicConfig.TOPIC_EXCHANGE,TopicConfig.ROUTING_WMS,map);
}
return ResponseEntity.noContent().build();
}
运行项目,观察swagger页面接口已经发生改变,已经有我们定义的接口了.
我们调用接口发送一条数据到队列中
可以看出已经把信息发送到队列中去了,传输的数据是cs2
我们调用swagger的接口是transferOrderTopic
,我们在通过交换机名称和路由关键字来查找队列的名称
我们在看TopicConfig配置类来查找对应的队列
找到对应的队列名称后
我们通过浏览器访问rabbitmq查看队列http://服务的ip:15672/
可以看出已经有一条信息在队列中因为我们没有写消费者,所以会堆积在队列中,点击队列的名称
到当前页面
可以看出数据是进行加密的
4.创建消费者
在Consumer包下创建TopicConsumer消费者创建2个方法来消费对应的对象信息!
@Component
public class TopicConsumer {
private static final Logger logger= LoggerFactory.getLogger(TopicConsumer.class);
@RabbitListener(queues = {"topic_order_message"})//队列名称(topic_order_message)
public void topicOrder(Map message){
System.out.println("topicOrder处理信息:"+message.toString());
System.out.println("处理订单完成:"+message.get("MESSAGE"));
}
@RabbitListener(queues = {"topic_wms_message"})//队列名称(topic_wms_message)
public void topicWms(Map message){
System.out.println("topicWms处理信息:"+message.toString());
System.out.println("处理订单完成:"+message.get("MESSAGE"));
}
}
运行程序,记得之前没有消费者我们就已经往队列里面添加了一条数据信息,现在我们启动的话应该会被消费掉在看一眼界面
运行完成后控制台已经打印消费信息了
可以看出信息从Rabbitmq获取出来后信息被解密成正常的数据了,而且不像我们之前在RabbiMq管理界面看到的密文一样.
在测试下仓库创建多条数据到队列中的情况一次传递的是订单对象
调用完成后可以看出入队列和消费队列的信息了
乍一看怎么感觉和Direct模式差不多,你是不是忘记了Topic可是具备转发功能的,而且Direct是点对点的传播,接下来测试转发功能!
5.Topic转发
前面说道消息是通过routing key与binding key的匹配关系进行路由的匹配分发那么改代码 ,现在我们尽量代码改动最小来展示转发,前面我们写了2个接口一个是订单的一个是wms的我们修改路由的关键字
改成以下
其他都不变,在重新运行程序调用订单的那个接口
可以看出存进去队列一条信息被被分发到了2个队列中然后,消费者把队列的数据进行消费
解释下
因为controller调用的方法使用的是TopicConfig的ROUTING_ORDER常量
字符串就是topic.routing.order
又因为我们定义了一个路由关键词叫 ROUTING_WMS
对应的字符串是"topic.routing.#"
,而且绑定的时候队列和关键字进行了绑定
导致路由关键字只要是#可以匹配多个词
之前的一样都会被转发到这个队列中
注意:的是由于我们队列是设置成了true持久化的导致后续测试
*代表一个的也触发了转发到对应的队列中,就算我们重启项目队列也依然存在,知道是裂开了后面登录Rabbitmq管理界面才发现这个问题改成false会打印一些提示信息不用管
在使用通配符*
匹配一个修改配置改成*
可以正常转发
那我把路由关键字在添加一个单词看看还能不能转发用
.隔开哦
可以看出没有被转发到另外一个队列中
四.Fanout Exchange模式
Fanout
模式下只要是绑定了Exchange(交换机)的Queue(队列)都会被接受到信息
1.前期准备
注意:为了测试Fanout 转发模式我们需要把Topic模式用到的类都删除,避免影响到Fanout 的测试,我这的做法是把Topic使用到的类@Configuration
和@Component
注释这样spring就不会加载这个对象
DirectConfig配置类注释上
TopicConsumer消费者类注释上
2.Fanout配置编写
没有创建路由关键字因为Fanout交换机必须要,只要绑定该交换机就都会分发数据到队列中
注意:在Spring中通过@Bean
创建对象时,如果多个bean的类型是一样的我们需要使用@Qualifier()
来指定使用的是那个@Bean,如果@Bean(name="cs")
没有指定name属性那么@Qualifier()里面填写的就是方法的名称,你可以在这个绑定交换机和队列的方法上看到参数有使用这个注解@Qualifier()
/**
* Fanout模式
*/
@Configuration
public class FanoutConfig {
/**
* 创建队列
*/
public static final String QUEUE_FANOUT_ORDER="fanout_order_message";//订单的队列
public static final String QUEUE_FANOUT_WMS="fanout_wms_message";//库存的队列
/**
* 配置交换机
*/
public static final String FANOUT_EXCHANGE="fanout";//topic类型交换机
/**
* 声明队列
* @return
*/
@Bean
public Queue fanoutQueueOrder(){
//持久 - 如果我们声明一个持久队列,则为真(该队列将在服务器重启后继续存在)
return new Queue(QUEUE_FANOUT_ORDER,false);
}
@Bean
public Queue fanoutQueueWms(){
//持久 - 如果我们声明一个持久队列,则为真(该队列将在服务器重启后继续存在)
return new Queue(QUEUE_FANOUT_WMS,false);
}
/**
* 声明交换机
* @return
*/
@Bean
public FanoutExchange FanoutExchange(){
//交换器名称、是否持久化、是否自动删除
return new FanoutExchange(FANOUT_EXCHANGE,true,false);
}
/***
* 队列绑定到交换机上
*/
@Bean
public Binding bindingExchangeOrder(@Qualifier("fanoutQueueWms") Queue queue, FanoutExchange exchange){
return BindingBuilder.bind(queue).to(exchange);
}
/***
* 队列绑定到交换机上
*/
@Bean
public Binding bindingExchangeWms(@Qualifier("fanoutQueueOrder") Queue queue, FanoutExchange exchange){
return BindingBuilder.bind(queue).to(exchange);
}
}
3.创建Fanout接口测试
RabbitMqController创建接口
生产者类不需要改动,我们只要把传人的数据改成FanoutConfig的交换机不需要传递路由关键词添加以下代码到RabbitMqController
类
@ApiOperation(value = "Fanout:订单信息")
@PostMapping("transferOrderFanout")
public ResponseEntity<Void> transferOrderFanout(@ApiParam(value = "数据信息",required = true) @RequestParam String message){
Map map=new HashMap();
map.put("UUID", UUID.randomUUID());
map.put("MESSAGE",message);
map.put("CREATEDATE", DateTime.now());//2021-12-21 10:35:31
//交换机名称 路由关键字 传递的数据
orderProducer.send(FanoutConfig.FANOUT_EXCHANGE,"",map);
return ResponseEntity.noContent().build();
}
4.创建消费者
在Consumer包下创建FanoutConsumer消费者创建2个方法来消费对应的对象信息!
@Component
public class FanoutConsumer {
private static final Logger logger= LoggerFactory.getLogger(FanoutConsumer.class);
@RabbitListener(queues = {"fanout_order_message"})//队列名称(fanout_order_message)
public void topicOrder(Map message){
System.out.println("fanoutOrder处理信息:"+message.toString());
System.out.println("处理订单完成:"+message.get("MESSAGE"));
}
@RabbitListener(queues = {"fanout_wms_message"})//队列名称(fanout_wms_message)
public void topicWms(Map message){
System.out.println("fanoutWms处理信息:"+message.toString());
System.out.println("处理订单完成:"+message.get("MESSAGE"));
}
}
运行项目查看浏览器RabbitMq的管理界面(里面有2个队列是Topic的先不管
)只看我们的Fanout的队列
调用swagger来运行接口测试
调用接口可以发现只要是绑定了FanoutExchange
交换机的队列都会收到信息
5.使用ApiPost请求接口
展示一下Apipost请求swagger的接口来调用队列,这个接口我们只要传递一个message测试和请求方式为Post
请求路径是通过url路径拼接起来的
这个就是请求路径
回到ApiPost上
可以看出已经调用上接口了,
五.补充知识点:
关于本次演示代码部分有个点需要补充下以免初学者产生疑问!
由于我在每种模式的配置类中都有创建对应的队列,这样就意味着启动项目就会往Rabbitmq中创建交换机,队列等操作
这样使用这种方式监听队列的信息是没问题的!
但是然后你没有创建队列
信息使用这种只填写队列名称来监听是不可行的!需要使用以下方式来解决这个问题
@RabbitListener(bindings = @QueueBinding(
value = @Queue(value = "topic_order_message",//队列名称
durable = "true"//是否持久保存
),
exchange = @Exchange(value = "topic_exchange",//交换机名称
ignoreDeclarationExceptions = "true",//忽略声明异常如果有这个交换机了就用mq里面的
type = ExchangeTypes.TOPIC, //交换机类型:direct,fanout,topic 等
durable = "true"//默认就是true 可以省略这个属性
),
key = {"direct.order.routing"}//路由关键字:routing key
))
public void topicOrder(Map message){
System.out.println("topicOrder处理信息:"+message.toString());
System.out.println("处理订单完成:"+message.get("MESSAGE"));
}
版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。
文章由极客之音整理,本文链接:https://www.bmabk.com/index.php/post/83853.html