本文主要是介绍RabbitMQ基础有这一篇就够了,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!
RabbitMQ基础篇
- 1. 同步异步
- 2. MQ技术选型
- 3. 数据隔离
- 4. SpringAMQP
- 5. WorkQueues模式
- 5.1 Work Queues
- 5.2 交换机(发布/订阅模式)
- 5.2.1 Fanout交换机(广播)
- 5.2.2 Direct交换机(定向)
- 5.2.3 Topic交换机(话题)
- 6. 声明队列、交换机以及进行绑定
- 6.1 基于API声明队列、交换机以及进行绑定
- 6.2 基于注解声明队列、交换机以及进行绑定
- 7. MQ消息转换器
- 问题提出
1. 同步异步
异步调用通常是基于消息通知的方式,包含三个角色:
- 消息发送者:投递消息的人,就是原来的调用者
- 消息接收者:接收和处理消息的人,就是原来的*服务提供者
- 消息代理:管理、暂存、转发消息,你可以把它理解成微信服务器
异步调用:
异调用的优势是什么?
- 耦合度低,拓展性强
- 异步调用,无需等待,
- 性能好 故障隔离,下游服务故障不影响上游业务
- 缓存消息,流量削峰填谷
异步调用的问题是什么?
- 不能立即得到调用结果,时效性差
- 不确定下游业务执行是否成功
- 业务安全依赖于Broker的可靠性
2. MQ技术选型
MQ (MessageQueue),中文是消息队列,字面来看就是存放消息的队列。也就是异步调用中的Broker。
可用性:系统正常运行的时间占总时间的比例。高可用性的消息队列系统能够在发生故障时快速恢复,保证数据的访问和处理不中断。
- RabbitMQ:使用集群和镜像队列可以提供较高的可用性,但需要额外的配置和管理。
- ActiveMQ:支持多种消息持久化模式,可以通过配置提高系统的可用性。
- Kafka:使用复制(Replication)机制来实现高可用性,数据会被复制到多个副本,通常至少三个。
单机吞吐量:指单个服务器节点在单位时间内可以处理的消息数量。
- RabbitMQ:吞吐量受节点内存和磁盘的限制,通常在消息量不是非常大的时候表现良好,每秒十万左右。
- ActiveMQ:吞吐量也受限于单机性能,但可以使用更高的配置来提高吞吐量。
- Kafka:设计上更倾向于高吞吐量,特别是对于批处理场景,可以支持每秒数百万的消息处理。
消息延迟:从消息发送到消息被消费者处理之间的时间长度。延迟会受到网络延迟、队列处理的负载以及消息本身处理时间的影响。
- RabbitMQ:支持低延迟的消息传递,但处理的消息量增加时,延迟可能会增加。
- ActiveMQ:延迟受限于消息队列的配置,合理配置可以保证较低的延迟。
- Kafka:除非进行特殊配置,默认情况下Kafka可能会有较大的延迟,因为它是面向批处理的。但在实际使用中可以通过调整配置来降低延迟。
消息可靠性:消息是否能够可靠地传递到消费者,以及消费者是否能够正确地处理消息。
- RabbitMQ:提供了多种消息确认机制(如事务、发布确认),可以确保消息被可靠地传递。
- ActiveMQ:同样提供了消息持久化、事务和消息确认等机制来确保消息的可靠性。K
- Kafka:通过复制机制来确保数据不会丢失,同时消费者可以通过ACK机制来确保消息的处理。
在实际选择消息队列技术时,需要根据项目的具体需求来决定。如果项目对消息的一致性和顺序传递要求高,同时消息量不大,可能倾向于选择 RabbitMQ 或 ActiveMQ。如果对数据传输效率和吞吐量有较高要求,并且容忍一定的消息延迟,Kafka 可能是更好的选择。
3. 数据隔离
对于小型企业而言,出于成本考虑,通常只会搭建一套MQ集群,公司内的多个不同项目同时使用(相当于一个MySQL的不同DB)。这个时候为了避免互相干扰, 会利用virtual host的隔离特性,将不同项目隔离。一般会做两件事情:
- 给每个项目创建独立的运维账号,将管理权限分离。
- 给每个项目创建不同的virtual host,将每个项目的数据隔离。
消息发送的注意事项有哪些?
- 交换机只能路由消息,无法存储消息
- 交换机只会路由消息,发送给与其绑定的队列,因此队列必须与交换机绑定
4. SpringAMQP
Advanced Message Queuing Protocol,是用于在应用程序之间传递业务消息的开放标准。该协议与语言和平台无关,更符合微服务中独立性的要求。
SpringAMQP收发消息的流程总结:
- 引入spring-boot-starter-amqp依赖
<!--AMQP依赖,包含RabbitMQ-->
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
- 配置rabbitmq服务端信息(consumer 和 publisher服务的application.yml中都需要添加配置)
spring:rabbitmq:host: 192.168.150.101 # 你的虚拟机IPport: 5672 # 端口virtual-host: /hmall # 虚拟主机username: hmall # 用户名password: 123 # 密码
- 利用RabbitTemplate发送消息
@SpringBootTest
public class SpringAmqpTest {@Autowiredprivate RabbitTemplate rabbitTemplate;@Testpublic void testSimpleQueue() {// 队列名称String queueName = "simple.queue";// 消息String message = "hello, spring amqp!";// 发送消息rabbitTemplate.convertAndSend(queueName, message);}
}
- 利用@RabbitListener注解声明要监听的队列,监听消息
package com.itheima.consumer.listener;import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Component
public class SpringRabbitListener {// 利用RabbitListener来声明要监听的队列信息// 将来一旦监听的队列中有了消息,就会推送给当前服务,调用当前方法,处理消息。// 可以看到方法体中接收的就是消息体的内容@RabbitListener(queues = "simple.queue")public void listenSimpleQueueMessage(String msg) throws InterruptedException {System.out.println("spring 消费者接收到消息:【" + msg + "】");}
}
5. WorkQueues模式
5.1 Work Queues
Work queues,任务模型。简单来说就是让多个消费者绑定到一个队列,共同消费队列中的消息。
消费者消息推送限制:
默认情况下,RabbitMQ的会将消息依次轮询投递给绑定在队列上的每一个消费者,假如有50条消息,两个消费者1,2,那么就会均分预先分给每个消费者25条消息。但这并没有考虑到消费者是否已经处理完消息,可能出现消息堆积。因此我们需要修改application.yml,设置preFetch值为1,确保同一时刻最多投递给消费者1条消息:
spring:rabbitmq:host: 192.168.150.101 # 你的虚拟机IPport: 5672 # 端口virtual-host: /hmall # 虚拟主机username: hmall # 用户名password: 123 # 密码listener: simple: prefetch: 1 # 每次只能获取一条消息,处理完成才能获取下一个消息
5.2 交换机(发布/订阅模式)
交换机的作用:
- 接收publisher发送的消息
- 将消息按照规则路由到与之绑定的队列
- Publisher:生产者,不再发送消息到队列中,而是发给交换机
- Exchange:交换机,一方面,接收生产者发送的消息。另一方面,知道如何处理消息,例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作,取决于Exchange的类型。
- Queue:消息队列也与以前一样,接收消息、缓存消息。不过队列一定要与交换机绑定。
- Consumer:消费者,与以前一样,订阅队列,没有变化
Exchange(交换机)只负责转发消息,不具备存储消息的能力,因此如果没有任何队列与Exchange绑定,或者没有符合路由规则的队列,那么消息会丢失!
交换机的类型有四种:
- Fanout:广播,将消息交给所有绑定到交换机的队列。最早在控制台使用的正是Fanout交换机
- Direct:订阅,基于RoutingKey(路由key)发送给订阅了消息的队列
- Topic:通配符订阅,与Direct类似,只不过RoutingKey可以使用通配符
- Headers:头匹配,基于MQ的消息头匹配,用的较少。
5.2.1 Fanout交换机(广播)
Fanout Exchange 会将接收到的消息路由到每一个跟其绑定的queue,所以也叫广播模式
利用SpringAMQP演示FanoutExchange的使用
需求如下:
- 在RabbitMQ控制台中,声明队列fanout.queue1和fanout.queue2
- 在RabbitMQ控制台中,声明交换机hmall.fanout,将两个队列与其绑定
- 在consumer服务中,编写两个消费者方法,分别监听fanout.queue1和fanout.queue2
- 在publisher中编写测试方法,向hmall.fanout发送消息
- 步骤1:在RabbitMQ控制台中,声明队列fanout.queue1和fanout.queue2
- 步骤2:在RabbitMQ控制台中,声明交换机hmall.fanout,将两个队列与其绑定
- 步骤3:在consumer服务中,编写两个消费者方法,分别监听fanout.queue1和fanout.queue2
@Test
public void testFanoutExchange() {// 交换机名称String exchangeName = "hmall.fanout";// 消息String message = "hello, everyone!";// 发送消息, 参数分别为:交换机名称, RoutingKey(暂时为空或者为null), 消息rabbitTemplate.convertAndSend(exchangeName, "", message);
}
- 步骤4:在publisher中编写测试方法,向hmall.fanout发送消息
@RabbitListener(queues = "fanout.queue1")
public void listenFanoutQueue1(String msg) {System.out.println("消费者1接收到Fanout消息:【" + msg + "】");
}@RabbitListener(queues = "fanout.queue2")
public void listenFanoutQueue2(String msg) {System.out.println("消费者2接收到Fanout消息:【" + msg + "】");
}
5.2.2 Direct交换机(定向)
Direct Exchange 会将接收到的消息根据规则路由到指定的Queue,因此称为定向路由。
- 每一个Queue都与Exchange设置一个BindingKey
- 发布者发送消息时,指定消息的RoutingKey
- Exchange将消息路由到BindingKey与消息RoutingKey一致的队列
描述下Direct交换机与Fanout交换机的差异?
- Fanout交换机将消息路由给每一个与之绑定的队列
- Direct交换机根据RoutingKey判断路由给哪个队列
- 如果多个队列具有相同RoutingKey,则与Fanout功能类似
利用SpringAMQP演示DirectExchange的使用
需求如下
- 在RabbitMQ控制台中,声明队列direct.queue1和direct.queue2
- 在RabbitMQ控制台中,声明交换机hmall. direct ,将两个队列与其绑定,direct.queue1的bindingKey为blud和red,direct.queue2的bindingKey为yellow和red
- 在consumer服务中,编写两个消费者方法,分别监听direct.queue1和direct.queue2
- 在publisher中编写测试方法,利用不同的RoutingKey向hmall. direct发送消息
-
步骤1:在RabbitMQ控制台中,声明队列direct.queue1和direct.queue2
-
步骤2:在RabbitMQ控制台中,声明交换机hmall. direct ,将两个队列与其绑定,direct.queue1的bindingKey为blud和red,direct.queue2的bindingKey为yellow和red
-
步骤3:在consumer服务中,编写两个消费者方法,分别监听direct.queue1和direct.queue2
@RabbitListener(queues = "direct.queue1")
public void listenDirectQueue1(String msg) {System.out.println("消费者1接收到direct.queue1的消息:【" + msg + "】");
}@RabbitListener(queues = "direct.queue2")
public void listenDirectQueue2(String msg) {System.out.println("消费者2接收到direct.queue2的消息:【" + msg + "】");
}
- 步骤4:在publisher中编写测试方法,利用不同的RoutingKey向hmall. direct发送消息
@Test
public void testSendDirectExchange() {// 交换机名称String exchangeName = "hmall.direct";// 消息String message = "红色警报!日本乱排核废水,导致海洋生物变异,惊现哥斯拉!";// 发送消息, RoutingKey为redrabbitTemplate.convertAndSend(exchangeName, "red", message);
}@Test
public void testSendDirectExchange() {// 交换机名称String exchangeName = "hmall.direct";// 消息String message = "最新报道,哥斯拉是居民自治巨型气球,虚惊一场!";// 发送消息, RoutingKey为bluerabbitTemplate.convertAndSend(exchangeName, "blue", message);
}
5.2.3 Topic交换机(话题)
TopicExchange与DirectExchange类似,区别在于routingKey可以是多个单词的列表,并且以 “.” 分割。
Queue与Exchange指定BindingKey时可以使用通配符:
- #:代指0个或多个单词
- *:代指一个单词
描述下Direct交换机与Topic交换机的差异?
- Topic交换机接收的消息RoutingKey可以是多个单词,以 . 分割
- Topic交换机与队列绑定时的bindingKey可以指定通配符
- #:代表0个或多个词
- *:代表1个词
利用SpringAMQP演示DirectExchange的使用
需求如下:
- 在RabbitMQ控制台中,声明队列topic.queue1和topic.queue2
- 在RabbitMQ控制台中,声明交换机hmall. topic ,将两个队列与其绑定
- 在consumer服务中,编写两个消费者方法,分别监听topic.queue1和topic.queue2
- 在publisher中编写测试方法,利用不同的RoutingKey向hmall. topic发送消息
步骤1:在RabbitMQ控制台中,声明队列topic.queue1和topic.queue2
步骤2:在RabbitMQ控制台中,声明交换机hmall. topic ,将两个队列与其绑定
步骤3:在consumer服务中,编写两个消费者方法,分别监听topic.queue1和topic.queue2
/*** topicExchange*/
@Test
public void testSendTopicExchange() {// 交换机名称String exchangeName = "hmall.topic";// 消息String message = "喜报!孙悟空大战哥斯拉,胜!";// 发送消息, RountingKey满足topic.queue1的Routing key:china.#rabbitTemplate.convertAndSend(exchangeName, "china.news", message);
}
步骤4:在publisher中编写测试方法,利用不同的RoutingKey向hmall. topic发送消息
@RabbitListener(queues = "topic.queue1")
public void listenTopicQueue1(String msg){System.out.println("消费者1接收到topic.queue1的消息:【" + msg + "】");
}@RabbitListener(queues = "topic.queue2")
public void listenTopicQueue2(String msg){System.out.println("消费者2接收到topic.queue2的消息:【" + msg + "】");
}
6. 声明队列、交换机以及进行绑定
SpringAMQP提供了API和注解两种方式来创建交换机、声明队列以及进行绑定;以后都不需要基于RabbitMQ控制台来创建队列、交换机进行创建,由程序启动时检查队列和交换机是否存在,如果不存在自动创建。
6.1 基于API声明队列、交换机以及进行绑定
SpringAMQP提供了几个类,用来声明队列、交换机及其绑定关系:
- Queue:用于声明队列,可以用工厂类QueueBuilder构建
- Exchange:用于声明交换机,可以用工厂类ExchangeBuilder构建
- Binding:用于声明队列和交换机的绑定关系,可以用工厂类BindingBuilder构建
SpringAMQP提供了一个Queue类,用来创建队列:
SpringAMQP还提供了一个Exchange接口,来表示所有不同类型的交换机:
而在绑定队列和交换机时,则需要使用BindingBuilder来创建Binding对象:
声明一个Fanout类型的交换机,并且创建队列与其绑定:
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类型交换机*/@Beanpublic FanoutExchange fanoutExchange(){// return ExchangeBuilder.fanoutExchange("hmall.fanout").build();return new FanoutExchange("hmall.fanout");}/*** 第1个队列*/@Beanpublic Queue fanoutQueue1(){// 来声明队列// QueueBuilder.durable("fanout.queue1").build();return new Queue("fanout.queue1");}/*** 绑定队列和交换机*/@Beanpublic Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);}/*** 第2个队列*/@Beanpublic Queue fanoutQueue2(){// 来声明队列// QueueBuilder.durable("fanout.queue2").build();return new Queue("fanout.queue2");}/*** 绑定队列和交换机*/@Beanpublic Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);}
}
声明一个direct模式类型的交换机,并且创建队列与其绑定:
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类型交换机*/@Beanpublic DirectExchange directExchange(){return ExchangeBuilder.directExchange("hmall.direct").build();}/*** 第1个队列*/@Beanpublic Queue directQueue1(){return new Queue("direct.queue1");}/*** 绑定队列和交换机*/@Beanpublic Binding bindingQueue1WithRed(Queue directQueue1, DirectExchange directExchange){return BindingBuilder.bind(directQueue1).to(directExchange).with("red");}/*** 绑定队列和交换机*/@Beanpublic Binding bindingQueue1WithBlue(Queue directQueue1, DirectExchange directExchange){return BindingBuilder.bind(directQueue1).to(directExchange).with("blue");}/*** 第2个队列*/@Beanpublic Queue directQueue2(){return new Queue("direct.queue2");}/*** 绑定队列和交换机*/@Beanpublic Binding bindingQueue2WithRed(Queue directQueue2, DirectExchange directExchange){return BindingBuilder.bind(directQueue2).to(directExchange).with("red");}/*** 绑定队列和交换机*/@Beanpublic Binding bindingQueue2WithYellow(Queue directQueue2, DirectExchange directExchange){return BindingBuilder.bind(directQueue2).to(directExchange).with("yellow");}
}
6.2 基于注解声明队列、交换机以及进行绑定
声明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 + "】");
}
声明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 + "】");
}
7. MQ消息转换器
Spring的消息发送代码接收的消息体是一个Object:
而在数据传输时,会把发送的消息序列化为字节发送给MQ,接收消息的时候,还会把字节反序列化为Java对象。默认情况下Spring采用的序列化方式是JDK序列化。JDK序列化存在下列问题:
- 数据体积过大
- 有安全漏洞
- 可读性差
发送消息后查看控制台:
可以看到控制台现实的消息格式不友好,可以引入Jackson依赖进行解决。
在publisher和consumer两个服务中都引入依赖:
<dependency><groupId>com.fasterxml.jackson.dataformat</groupId><artifactId>jackson-dataformat-xml</artifactId><version>2.9.10</version>
</dependency>
配置消息转换器,在publisher和consumer两个服务的启动类中添加一个Bean即可()。消息转换器中添加的messageId可以便于我们将来做幂等性判断。
@Bean
public MessageConverter messageConverter(){// 1.定义消息转换器Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();// 2.配置自动创建消息id,用于识别不同消息,也可以在业务中基于ID判断是否是重复消息jackson2JsonMessageConverter.setCreateMessageIds(true);return jackson2JsonMessageConverter;
}
问题提出
- 在6.1中使用API声明队列、交换机以及确定绑定关系。Direct交换机中如何确定绑定关系的,direct.queue1绑定hmall.direct,rountingKey为什么是red,bule,而不是red,yellow???
- 参数传递顺序?
这篇关于RabbitMQ基础有这一篇就够了的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!