一、消息队列MQ概述
MQ全称为Message Queue,消息队列是应用程序和应用程序之间的通信方法。
1.1 为什么使用MQ
在项目中,可将一些无需即时返回且耗时的操作提取出来,进行异步处理,而这种异步处理的方式大大的节省了服务器的请求响应时间,从而提高了系统的吞吐量。
开发中消息队列通常有如下优点:
**(1) 异步提速: **异步处理任务,将不需要同步处理且耗时长的操作交由消息队列进行异步处理,大大提高了应用程序的处理时间。
**(2) 应用解耦: **应用程序解耦合,MQ充当中介,生产方通过MQ与消费方交互,它将应用程序解耦合了
**(3) 削峰填谷: **在访问量剧增的情况下,应用仍然需要继续发挥作用,但是这样的突发流量并不常见;如果为以能处理这类峰值访问为标准,来投入资源随时待命无疑是巨大的浪费。使用MQ能够使关键组件顶住突发的访问压力,而不会因为突发的超负荷请求而完全崩溃。
**(4) 可恢复性: **系统的一部分组件失效时,不会影响到整个系统。MQ降低了程序间的耦合度,所以即使一个处理消息的挂掉,加入队列中的消息仍然可以在系统恢复后被再次被改程序处理。
(5) 排序保证: 消息队列可以控制数据处理的顺序,因为消息队列本身使用的是队列这个数据结构,
FIFO
(先进先出),在一些场景中,数据处理的顺序很重要,比如商品下单、抢票、秒杀…等。
1.2. 消息队列产品
市场上常见的消息队列有如下:
1.3. AMQP 和 JMS
Dubbo协议:Dubbo 缺省协议采用单一长连接和 NIO 异步通讯,适合于小数据量大并发的服务调用,以及服务消费者机器数远大于服务提供者机器数的情况。
HTTP协议(HyperText Transfer Protocol,超文本传输协议)是因特网上应用最为广泛的一种网络传输协议,所有的WWW文件都必须遵守这个标准。
MQ是消息通信的模型;实现MQ的大致有两种主流方式:AMQP、JMS。
1.3.1. AMQP
AMQP,即Advanced Message Queuing Protocol,一个提供统一消息服务的应用层标准高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计。
1.3.2. JMS
JMS即Java消息服务(JavaMessage Service)应用程序接口,是一个Java平台中关于面向消息中间件的API,用于在两个应用程序之间,或分布式系统中发送消息,进行异步通信。
1.3.3. AMQP 与 JMS 区别
JMS是定义了统一的接口,来对消息操作进行统一;AMQP是通过规定协议来统一数据交互的格式JMS限定了必须使用Java语言;AMQP只是协议,不规定实现方式,因此是跨语言的。JMS规定了两种消息模式;而AMQP的消息模式更加丰富.
JMSAMQP定义Java apiWire-protocol跨语言否是跨平台否是
1.4. RabbitMQ
RabbitMQ是由erlang语言开发,基于AMQP(Advanced Message Queue 高级消息队列协议)协议实现的消息队列产品,它是一种应用程序之间的通信方法,消息队列在分布式系统开发中应用非常广泛。
RabbitMQ官方地址:http://www.rabbitmq.com/
RabbitMQ提供了6种模式:Hello Word简单模式,work工作模式,Publish/Subscribe发布与订阅模式,Routing路由模式,Topics主题模式(通配符模式),RPC远程调用模式(远程调用,不太算MQ;不作介绍)
官网对应模式介绍:https://www.rabbitmq.com/getstarted.html
应用场景:
1、双十一商品秒杀/抢票功能实现
我们在双11的时候,当我们凌晨大量的秒杀和抢购商品,然后去结算的时候,就会发现,界面会提醒我们,让我们稍等,以及一些友好的图片文字提醒。而不是像前几年的时代,动不动就页面卡死,报错等来呈现给用户。
2、积分兑换(积分可用于多平台)
积分兑换模块,有一个公司多个部门都要用到这个模块,这时候就可以通过消息队列解耦这个特性来实现。 各部门系统做各部门的事,但是他们都可以用这个积分系统进行商品的兑换等。其他模块与积分模块完全解耦。
3、大平台用户注册
发送邮件、用户大数据分析操作等 基于同步变异步功能实现
用户注册真实操作步骤:
- 用户注册选择的兴趣标签,根据用户的属性,行为进行用户分析,计算出推荐内容
- 注册后可能需要发送邮件给用户
- 发送短信给用户
- 发送给用户指南的系统通知
- …等等
正常情况注册,不出现高并发,假如有大量的用户注册,发生了高并发,就会出现如下情况:
邮件接口承受不住,或是分析信息时的大量计算使 cpu 满载。这将会出现,虽然用户数据记录很快的添加到数据库中了,但是执行流程却卡在发邮件或分析用户信息的情况。导致请求的响应时间大幅增长,甚至出现超时,这就有点不划算了。面对这种情况一般也是将这些操作放入消息队列(生产者消费者模型),消息队列慢慢的进行处理,同时可以很快的完成注册请求,不会影响用户使用其他功能。
1.5 相关定义:
- Connection:publisher/consumer 和 broker 之间的 TCP 连接
- Channel: 消息通道,在客户端的每个连接里,可建立多个channel,每个channel代表一个会话任务
- Exchange: 消息交换机,它指定消息按什么规则,路由到哪个队列
- Queue: 消息队列载体,每个消息都会被投入到一个或多个队列
- VHost: 虚拟主机,一个broker里可以开设多个vhost,用作不同用户的权限分离。
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-X9lFeYaO-1646011259625)(RabbitMQ%E8%AE%B2%E4%B9%89.assets/image-20191226053108367.png)]
由Exchange、Queue、RoutingKey三个才能决定一个消息从Exchange到Queue的唯一的线路。
二、安装及配置RabbitMQ
2.1. docker中安装RabbitMQ
- 下载镜像
docker pull rabbitmq:management
- 创建容器
docker run -di --name=my_rabbitmq -p 5671:5671 -p 5672:5672 -p4369:4369 -p 15671:15671 -p 15672:15672 -p 25672:25672 rabbitmq:management
解释如下:
15672 (if management plugin is enabled.管理界面 )
15671 management监听端口
5672, 5671 (AMQP 0-9-1 without and with TLS 消息队列协议是一个消息协议)
4369 (epmd) epmd 代表 Erlang 端口映射守护进程
25672 (Erlang distribution)
- 访问后台
浏览器中输入地址
http://192.168.200.128:15672/
- 设置容器开机自动启动
docker update --restart=always 容器ID
2.2. 用户以及Virtual Hosts配置
2.2.1. 用户角色
RabbitMQ在安装好后,可以访问http://192.168.200.128:15672;其自带了guest/guest的用户名和密码;如果需要创建自定义用户;那么也可以登录管理界面后,如下操作:
角色说明:
- 超级管理员(administrator):可登陆管理控制台,可查看所有的信息,并且可以对用户,策略(policy)进行操作。
- 监控者(monitoring):可登陆管理控制台,同时可以查看rabbitmq节点的相关信息(进程数,内存使用情况,磁盘使用情况等)
- 策略制定者(policymaker):可登陆管理控制台, 同时可以对策略(policy)进行管理。但无法查看节点及其相关信息(上图红框标识的部分)。
- 普通管理者(management):仅可登陆管理控制台,无法看到节点信息,也无法对策略进行管理。
- 其他无法登陆管理控制台,通常就是普通的生产者和消费者。
2.2.2. Virtual Hosts配置
RabbitMQ的权限管理;在RabbitMQ中可以虚拟消息服务器Virtual Host,每个Virtual Hosts相当于一个相对独立的RabbitMQ服务器,每个VirtualHost之间是相互隔离的。exchange、queue、message不能互通。Virtual Name一般以/开头。
- 创建Virtual Hosts
- 设置Virtual Hosts权限
2.2.3 添加队列
持久化:如果选durable,则队列消息自动持久化到磁盘上,如果选transient,则不会持久化;
自动删除:默认值no,如果yes,则在消息队列没有使用的情况下,队列自行删除。
2.2.4 添加交换机
自动删除:默认值no,如果是yes,则在将所有队列与交换机取消绑定之后,交换机将自动删除。
交换机类型:
- fanout:广播类型
- direct:路由类型
- topic:通配符类型,基于消息的路由键路由
- headers:通配符类型,基于消息的header路由
内部交换器:默认值no,如果是yes,消息无法直接发送到该交换机,必须通过交换机的转发才能到达次交换机。本交换机只能与交换机绑定。
2.2.5 队列与交换机进行绑定
三、Spring Boot整合RabbitMQ
3.1. 简介
在spring boot项目中,只需要引入start-amqp起步依赖,即可整合RabbitMQ成功;我们基于SpringBoot封装的RabbitTemplate模板对象,可以非常方便的发送消息,接收消息(使用注解)。
amqp的官方GitHub地址:https://github.com/spring-projects/spring-amqp
一般在开发过程中,我们有两个角色:
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-LDbo5j8z-1646011259630)(RabbitMQ%E8%AE%B2%E4%B9%89.assets/image-20191224123350539.png)]
3.2. 搭建步骤:
1、创建父工程:
2、生产者工程:
- 创建SpringBoot工程:rabbitmq-producer
- 勾选起步依赖坐标:spring for RabbitMQ
- 配置RabbitMQ:服务host地址及端口、虚拟主机、服务账户密码
3、消费者工程:
- 创建SpringBoot工程:rabbitmq-consumer
- 勾选起步依赖坐标:spring for RabbitMQ
- 配置RabbitMQ:服务host地址及端口、虚拟主机、服务账户密码
3.3. 搭建过程:
3.3.1 创建父工程:
创建生产者工程:rabbitmq-producer(创建过程不在赘述)
创建消费者工程:rabbitmq-consumer(创建过程不在赘述)
3.3.2 搭建生产者工程
1、创建工程
创建SpringBoot的生产者工程:rabbitmq-producer
2、起步依赖坐标
pom.xml文件内容为如下:
<?xml version="1.0" encoding="UTF-8"?><projectxmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.6.3</version><relativePath/><!-- lookup parent from repository --></parent><groupId>com.example</groupId><artifactId>rabbitmq-producer</artifactId><version>0.0.1-SNAPSHOT</version><name>rabbitmq-producer</name><description>Demo project for Spring Boot</description><properties><java.version>1.8</java.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-devtools</artifactId><scope>runtime</scope><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit-test</artifactId><scope>test</scope></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId><scope>test</scope></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build><repositories><repository><id>spring-milestones</id><name>Spring Milestones</name><url>https://repo.spring.io/milestone</url><snapshots><enabled>false</enabled></snapshots></repository><repository><id>spring-snapshots</id><name>Spring Snapshots</name><url>https://repo.spring.io/snapshot</url><releases><enabled>false</enabled></releases></repository></repositories><pluginRepositories><pluginRepository><id>spring-milestones</id><name>Spring Milestones</name><url>https://repo.spring.io/milestone</url><snapshots><enabled>false</enabled></snapshots></pluginRepository><pluginRepository><id>spring-snapshots</id><name>Spring Snapshots</name><url>https://repo.spring.io/snapshot</url><releases><enabled>false</enabled></releases></pluginRepository></pluginRepositories></project>
3、启动类
@SpringBootApplicationpublicclassSpringbootRabbitmqProducerApplication{publicstaticvoidmain(String[] args){SpringApplication.run(SpringbootRabbitmqProducerApplication.class, args);}}
4、配置RabbitMQ
1)配置文件application.properties,内容如下:
# RabbitMQ 服务host地址
spring.rabbitmq.host=192.168.200.128
# 端口
spring.rabbitmq.port=5672
# 虚拟主机地址
spring.rabbitmq.virtual-host=/wwg
# rabbit服务的用户名
spring.rabbitmq.username=admin
# rabbit服务的密码
spring.rabbitmq.password=admin
3.3.3 搭建消费者工程
1、创建工程
创建SpringBoot的消费者工程:rabbitmq-consumer
2、勾选起步依赖坐标
pom.xml文件内容为如下:
<?xml version="1.0" encoding="UTF-8"?><projectxmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.6.3</version><relativePath/><!-- lookup parent from repository --></parent><groupId>com.example</groupId><artifactId>rabbitmq-consumer</artifactId><version>0.0.1-SNAPSHOT</version><name>rabbitmq-consumer</name><description>Demo project for Spring Boot</description><properties><java.version>1.8</java.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-devtools</artifactId><scope>runtime</scope><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.76</version></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build><repositories><repository><id>spring-milestones</id><name>Spring Milestones</name><url>https://repo.spring.io/milestone</url><snapshots><enabled>false</enabled></snapshots></repository><repository><id>spring-snapshots</id><name>Spring Snapshots</name><url>https://repo.spring.io/snapshot</url><releases><enabled>false</enabled></releases></repository></repositories><pluginRepositories><pluginRepository><id>spring-milestones</id><name>Spring Milestones</name><url>https://repo.spring.io/milestone</url><snapshots><enabled>false</enabled></snapshots></pluginRepository><pluginRepository><id>spring-snapshots</id><name>Spring Snapshots</name><url>https://repo.spring.io/snapshot</url><releases><enabled>false</enabled></releases></pluginRepository></pluginRepositories></project>
3、启动类
@SpringBootApplicationpublicclassSpringbootRabbitmqConsumerApplication{publicstaticvoidmain(String[] args){SpringApplication.run(SpringbootRabbitmqConsumerApplication.class, args);}}
4、配置RabbitMQ
application.properties,内容如下:
# RabbitMQ 服务host地址
spring.rabbitmq.host=192.168.200.128
# 端口
spring.rabbitmq.port=5672
# 虚拟主机地址
spring.rabbitmq.virtual-host=/wwg
# rabbit服务的用户名
spring.rabbitmq.username=admin
# rabbit服务的密码
spring.rabbitmq.password=admin
四、RabbitMQ五种工作模式【重要】
4.1 Hello World简单模式
4.1.1 什么是简单模式
4.1.2 RabbitMQ管理界面操作
- 创建simple_queue队列用于演示Hello World简单模式
- 点击
simple_queue
可以进入到这个queue的管理界面 - 点击
Get Message
按钮可以获取查看队列中的消息
4.1.3 生产者代码
- rabbitmq-producer项目测试代码如下:
packagecom.sg.test;importorg.junit.Test;importorg.junit.runner.RunWith;importorg.springframework.amqp.rabbit.core.RabbitTemplate;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.boot.test.context.SpringBootTest;importorg.springframework.test.context.junit4.SpringRunner;@RunWith(SpringRunner.class)@SpringBootTestpublicclassDemo01TestSimpleQueue{@AutowiredprivateRabbitTemplate rabbitTemplate;@Test//测试简单模式publicvoidtestSimple(){//向消息队列发送一条简单消息/**
* 参数1:消息队列名称
* 参数2:消息内容
*/
rabbitTemplate.convertAndSend("queue1","hello !");}}
4.1.4 消费者代码
- rabbitmq-consumer项目创建监听器:
packagecom.sg.rabbitmq.simple;importorg.springframework.amqp.rabbit.annotation.RabbitHandler;importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.stereotype.Component;/**
* 消费者,接收消息队列消息监听器
* 必须将当前监听器对象注入Spring的容器中
*/@ComponentpublicclassSimpleListener{@RabbitListener(queues ="queue1")publicvoidsimpleHandler(String msg){System.out.println("=====接收消息====>"+msg);}}
然后启动SpringbootRabbitmqConsumerApplication, 就可以接收到RabbitMQ服务器发送来的消息
4.2 Work queues工作队列模式
4.2.1 什么是工作队列模式
4.2.2 RabbitMQ管理界面操作
- 创建
work_queue
队列用于演示work工作队列模式
4.2.3 生产者代码
rabbitmq-producer项目测试代码如下:
packagecom.sg.test;importorg.junit.Test;importorg.junit.runner.RunWith;importorg.springframework.amqp.rabbit.core.RabbitTemplate;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.boot.test.context.SpringBootTest;importorg.springframework.test.context.junit4.SpringRunner;@RunWith(SpringRunner.class)@SpringBootTestpublicclassDemo02TestWorkQueue{@AutowiredprivateRabbitTemplate rabbitTemplate;@Test//测试工作队列模式(一个生产者多个消费者) 测试结果: 每条消息只能被一个消费者消费(竞争关系)publicvoidtestWorkQueue(){for(int i =0; i <1000; i++){
rabbitTemplate.convertAndSend("work_queue","hello 【"+ i +"】");}}}
4.2.4 消费者代码
- rabbitmq-consumer项目创建监听器1:
packagecom.sg.rabbitmq.work;importorg.springframework.amqp.rabbit.annotation.RabbitHandler;importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.stereotype.Component;@ComponentpublicclassWorkListener1{@RabbitListener(queues ="work_queue")publicvoidwork_queue1(String msg,Channel channel,Message message)throwsIOException{System.out.println("=====接收消息1====>"+msg);
channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);}@RabbitListener(queues ="work_queue")publicvoidwork_queue2(String msg){System.out.println("=====接收消息2====>"+msg);}}
注意:两个消费者为竞争关系,消费者1收到的消息消费者2不会收到
4.3 三种模式概览
而在订阅模型中,多了一个exchange角色,而且过程略有变化:
- P:生产者,也就是要发送消息的程序,但是不再发送到队列中,而是发给X(交换机)
- C:消费者,消息的接受者,会一直等待消息到来。
- Queue:消息队列,接收消息、缓存消息。
- Exchange:交换机,图中的X。一方面,接收生产者发送的消息。另一方面,知道如何处理消息,例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作,取决于Exchange的类型。
Exchange有常见以下3种类型:
- Fanout:广播 将消息交给所有绑定到交换机的队列,没有路由键。消息队列需要绑定交换机。fanout 类型交换机转发消息是最快的。广播模式说的就是Fanout类型交换机
- Direct:定向 把消息定向交给消息队列,通过路由键(routing key)指定。消息队列需要绑定交换机。要求,发送消息的路由键与交换机的路由键完全匹配。路由模式说的就是 direct 类型交换机。
- Topic:主题(通配符) 把消息通过通配方式交给消息队列,通过路由键(routing key)指定。消息队列需要绑定交换机。要求发送消息的路由键满足交换机路由键的通配符条件。通配符模式说的就是topic模式交换机。 - 其中通配符模式使用了2个通配符,
#
号和*
号。- 符号#
匹配一个或多个单词,用.分隔- 符号*
匹配不多不少一个单词- 因此audit.#
能够匹配到audit.irs.corporate
,但是audit.*
只会匹配到audit.irs
。
注意:Exchange(交换机)只负责转发消息,不具备存储消息的能力,因此如果没有任何队列与Exchange绑定,或者没有符合路由规则的队列,那么消息会丢失
4.4 Publish/Subscribe发布与订阅模式
4.4.1 什么是发布订阅模式
4.4.2 RabbitMQ管理界面操作
- 创建两个队列 fanout_queue1和 fanout_queue2(创建步骤请看上面)
- 创建Exchange交换器
fanout_exchange
(创建步骤请看上面) - 将创建的fanout_exchange交换器和 fanout_queue1, fanout_queue2队列绑定(绑定步骤请看上面)
4.4.3 生产者代码
- rabbitmq-producer项目测试代码如下:
packagecom.sg.test;importorg.junit.Test;importorg.junit.runner.RunWith;importorg.springframework.amqp.rabbit.core.RabbitTemplate;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.boot.test.context.SpringBootTest;importorg.springframework.test.context.junit4.SpringRunner;/**
* 目标:将消息发送给交换机,通过交换机广播给消息队列,路由键为空字符串
*/@RunWith(SpringRunner.class)@SpringBootTestpublicclassDemo03TestPublishAndSubscribe{@AutowiredprivateRabbitTemplate rabbitTemplate;@Test//测试广播方式 测试结果: 两个消息队列均会接收全部消息publicvoidtestFanout(){for(int i =0; i <100; i++){/**
* 参数1:交换机名称
* 参数2:路由键
* 参数3:消息内容
*/
rabbitTemplate.convertAndSend("fanout_ex","","hello 【"+ i +"】!");}}}
4.4.4 消费者代码
- rabbitmq-consumer项目创建监听器:
packagecom.sg.rabbitmq.pubandsub;importorg.springframework.amqp.rabbit.annotation.RabbitHandler;importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.stereotype.Component;@ComponentpublicclassPubAndSubListener1{@RabbitListener(queues ="fanout_queue1")publicvoidfanout1(String msg,Channel channel,Message message)throwsIOException{System.out.println("=====fanout_queue1====>"+msg);
channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);}@RabbitListener(queues ="fanout_queue2")publicvoidfanout2(String msg){System.out.println("=====fanout_queue2====>"+msg);}}
结果:fanout_queue1和fanout_queue2均会收到相同数据
4.5 Routing路由模式
4.5.1 什么是路由模式
4.5.2 RabbitMQ管理界面操作
创建两个队列分别叫做
routing_queue1
和
routing_queue2
创建交换器
routing_exchange
, 类型为
direct
, 用于演示路由模式
设置绑定: 将创建的交换器
routing_exchange
和
routing_queue1
,
routing_queue2
绑定在一起, 路由键Routing Key分别为
info
和
error
;
4.5.3 生产者代码
- rabbitmq-producer项目测试代码如下:
packagecom.sg.test;importorg.junit.Test;importorg.junit.runner.RunWith;importorg.springframework.amqp.rabbit.core.RabbitTemplate;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.boot.test.context.SpringBootTest;importorg.springframework.test.context.junit4.SpringRunner;/**
* 目标:将消息发送给交换机,通过交换机路由给指定的消息队列,通过路由键类指定
*/@RunWith(SpringRunner.class)@SpringBootTestpublicclassDemo04TestRoutingModel{@AutowiredprivateRabbitTemplate rabbitTemplate;@Test//测试路由方式 测试结果: 根据routingKey的设置,不同的消费者进行消费publicvoidtestDirect(){for(int i =0; i <100; i++){if(i %2==0){
rabbitTemplate.convertAndSend("routing_ex","info","hello 【"+ i +"】!");}else{
rabbitTemplate.convertAndSend("routing_ex","error","hello 【"+ i +"】!");}}}}
4.5.4 消费者代码
- rabbitmq-consumer项目创建监听器:
packagecom.sg.rabbitmq.routing;importorg.springframework.amqp.rabbit.annotation.RabbitHandler;importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.stereotype.Component;/**
* 路由模式:消息队列接收监听器1,接收来自路由模式发送的消息
*/@Component@RabbitListener(queues ="routing_queue1")publicclassRoutingListener1{@RabbitHandlerpublicvoidroutingHandler(String msg){System.out.println("=====路由模式消息接收监听器【1】=====>"+msg);}}
- rabbitmq-consumer项目创建监听器:
packagecom.sg.rabbitmq.routing;importorg.springframework.amqp.rabbit.annotation.RabbitHandler;importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.stereotype.Component;/**
* 路由模式:消息队列接收监听器2,接收来自路由模式发送的消息
*/@ComponentpublicclassRoutingListener2{@RabbitListener(queues ="routing_queue2")publicvoidroutingInfo(String msg,Channel channel,Message message)throwsIOException{System.out.println("=====routingInfo====>"+msg);
channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);}@RabbitListener(queues ="routing_queue1")publicvoidroutingError(String msg){System.out.println("=====routingError====>"+msg);}}
结果:根据routing_key消息被不同消费者消费
4.6 Topics通配符模式(主题模式)
4.6.1 什么是通配符(主题)模式
Topic
类型与
Direct
相比,都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key的时候使用通配符!
Routingkey
: 一般都是有一个或多个单词组成,多个单词之间以”.”分割,例如:
item.insert
通配符规则:
#:匹配一个或多个词,多个词用点号分隔
*:匹配不多不少恰好1个词
举例:
item.#: 能够匹配
item.insert.abc.bbc
或者
item.insert
**item.*:**只能匹配
item.insert
4.6.2 RabbitMQ管理界面操作
创建队列
topic_queue1
和
topic_queue1
!
创建交换器
topic_exchange
, type类型为
topic
- 设置绑定:
topic_queue1
绑定的Routing Key路由键为item.*
topic_queue2
绑定的Routing Key路由键为item.#
4.6.3 生产者代码
- rabbitmq-producer项目测试代码如下:
packagecom.sg.test;importorg.junit.Test;importorg.junit.runner.RunWith;importorg.springframework.amqp.rabbit.core.RabbitTemplate;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.boot.test.context.SpringBootTest;importorg.springframework.test.context.junit4.SpringRunner;/**
* 目标:将消息发送给交换机,通过交换机路由给指定的消息队列,路由键使用通配符
*/@RunWith(SpringRunner.class)@SpringBootTestpublicclassDemo05TestTopicModel{@AutowiredprivateRabbitTemplate rabbitTemplate;@Test//测试Topic方式publicvoidtestTopic(){//向通配符交换机发送消息
rabbitTemplate.convertAndSend("topic_ex","item.insert","hello 路由键item.insert");
rabbitTemplate.convertAndSend("topic_ex","item.insert.abc","hello 路由键:item.insert.abc");}}
4.6.4 消费者代码
- rabbitmq-consumer项目创建监听器:
packagecom.sg.rabbitmq.topic;importorg.springframework.amqp.rabbit.annotation.RabbitHandler;importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.stereotype.Component;/**
* 通配符模式:消息队列接收监听器1,接收来自通配符模式发送的消息
*
*/@Component@RabbitListener(queues ="topic_queue1")publicclassTopicListener1{@RabbitListener(queues ="topic_queue1")publicvoidtopic1(String msg){System.out.println("=====topic*====>"+msg);}@RabbitListener(queues ="topic_queue2")publicvoidtopic2(String msg){System.out.println("=====topic#====>"+msg);}}
结果:
item.#: 能够匹配
item.insert.abc.bbc
或者
item.insert
**item.*:**只能匹配
item.insert
## 4.7 模式总结RabbitMQ
工作模式:
- **简单模式 HelloWorld : **
- 角色:一个生产者、一个队列、一个消费者,不需要交换机(默认交换机)
- **工作队列模式 Work Queue:**
- 角色:一个生产者、多个队列、多个消费者(竞争关系),不需要交换机(默认交换机)
- **发布订阅模式 Publish/subscribe: **
- 角色:一个生产者、一个交换机、多个队列、多个消费者
- 交换机类型为**==fanout==**,并且交换机和队列要进行绑定,不设置路由键(routing key)
- 当发送消息到交换机后,交换机会将消息广播发送到绑定的队列
- **路由模式 Routing: **
- 角色:一个生产者、一个交换机、多个队列、多个消费者
- 交换机类型为**==direct==**,并且交换机和队列要进行绑定,并且需要设置路由键(routing key)
- 当发送消息到交换机后,交换机会根据路由键(routing key)将消息发送到队列
- **通配符模式 Topic: **
- 角色:一个生产者、一个交换机、多个队列、多个消费者
- 交换机类型为**==topic==**,并且交换机和队列要进行绑定,并且需要设置路由键(routing key)
- 路由键字符串使用通配符,`*`和`#`,*好匹配一个单词,#匹配一个或多个单词。单词用`.`分隔
- 当发送消息到交换机后,交换机会根据路由键(routing key)加通配符将消息发送到队列
4.7 模式总结RabbitMQ
工作模式:
- **简单模式 HelloWorld : ** - 角色:一个生产者、一个队列、一个消费者,不需要交换机(默认交换机)
- 工作队列模式 Work Queue:- 角色:一个生产者、多个队列、多个消费者(竞争关系),不需要交换机(默认交换机)
- 发布订阅模式 Publish/subscribe: ** - 角色:一个生产者、一个交换机、多个队列、多个消费者- 交换机类型为fanout**,并且交换机和队列要进行绑定,不设置路由键(routing key)- 当发送消息到交换机后,交换机会将消息广播发送到绑定的队列
- 路由模式 Routing: ** - 角色:一个生产者、一个交换机、多个队列、多个消费者- 交换机类型为direct**,并且交换机和队列要进行绑定,并且需要设置路由键(routing key)- 当发送消息到交换机后,交换机会根据路由键(routing key)将消息发送到队列
- 通配符模式 Topic: ** - 角色:一个生产者、一个交换机、多个队列、多个消费者- 交换机类型为topic*,并且交换机和队列要进行绑定,并且需要设置路由键(routing key)- 路由键字符串使用通配符,
*
和#
,好匹配一个单词,#匹配一个或多个单词。单词用.
分隔- 当发送消息到交换机后,交换机会根据路由键(routing key)加通配符将消息发送到队列
第一次写博客希望对大家有所帮助。
版权归原作者 呜昂王0521 所有, 如有侵权,请联系我们删除。