一、同步通讯的优缺点
- 时效性较强,可以立即得到结果
- 但是耦合度高,性能和吞吐能力下降
- 有额外的资源消耗
二、异步通讯的优缺点
- 耦合度低,吞吐量提升
- 故障隔离,流量削峰
- 依赖于Broker的可靠性、安全性和吞吐能力
三、什么是MQ
- MQ(MessageQueue),指消息队列,也就是事件驱动架构中的Broker
- 常见的有RabbitMQ、ActiveMQ、RocketMQ、Kafka等
四、RabbitMQ入门
- 架构如下
- 常见概念如下
- channel:信道,操作MQ的工具
- exchange:交换器,路由消息到队列中
- queue:队列,缓存消息
- virtual host:虚拟主机,对queue、exchange等资源进行逻辑分组
- Publisher:生产者,生产消息的服务,将消息发送到queue中
- Consumer:消费者,消费消息的服务,监听queue,拉取消息
五、常见消息模型
- 基本消息队列(BaseQueue)
- 工作消息队列(WorkQueue)
- 发布订阅(Publish、Subscribe)
- Fanout Exchange:广播
- Direct Exchange:路由
- Topic Exchange:主题
- Fanout Exchange:广播
六、简单队列模型(一对一)
- 三个基本角色
- publisher:消息发布者,将消息发送到队列queue中
- queue:消息队列,负责接受并缓存消息
- consumer:订阅队列,处理队列中的消息
- 消息发送流程:
- 建立connection
- 创建channel
- 利用channel声明队列
- 利用channel向队列发送消息
- 消息接收流程:
- 建立connection
- 创建channel
- 利用channel声明队列
- 定义consumer的消费行为handleDelivery()
- 利用channel将消费者与队列绑定
- 代码如下
//Publisher
public void testSendMessage() throws IOException, TimeoutException {
// 1.建立连接
ConnectionFactory factory = new ConnectionFactory();
// 1.1.设置连接参数,分别是:主机名、端口号、vhost、用户名、密码
factory.setHost("192.168.150.101");
factory.setPort(5672);
factory.setVirtualHost("/");
factory.setUsername("itcast");
factory.setPassword("123321");
// 1.2.建立连接
Connection connection = factory.newConnection();
// 2.创建通道Channel
Channel channel = connection.createChannel();
// 3.创建队列
String queueName = "simple.queue";
channel.queueDeclare(queueName, false, false, false, null);
// 4.发送消息
String message = "hello, rabbitmq!";
channel.basicPublish("", queueName, null, message.getBytes());
System.out.println("发送消息成功:【" + message + "】");
// 5.关闭通道和连接
channel.close();
connection.close();
}
//Consumer
public static void main(String[] args) throws IOException, TimeoutException {
// 1.建立连接
ConnectionFactory factory = new ConnectionFactory();
// 1.1.设置连接参数,分别是:主机名、端口号、vhost、用户名、密码
factory.setHost("192.168.124.150");
factory.setPort(5672);
factory.setVirtualHost("/");
factory.setUsername("itcast");
factory.setPassword("123321");
// 1.2.建立连接
Connection connection = factory.newConnection();
// 2.创建通道Channel
Channel channel = connection.createChannel();
// 3.创建队列
String queueName = "simple.queue";
channel.queueDeclare(queueName, false, false, false, null);
// 4.订阅消息
channel.basicConsume(queueName, true, new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
// 5.处理消息
String message = new String(body);
System.out.println("接收到消息:【" + message + "】");
}
});
System.out.println("等待接收消息。。。。");
}
七、什么是SpringAMQP
- AMQP:是用于在应用程序或之间传递业务消息的开发标准,该协议与语言与平台无关,更符合微服务中独立性的要求
- SpringAMQP:是基于AMQP协议定义的一套API规范,提供了模板来发送和接收消息
八、使用SpringAMQP的简单步骤
- 引入AMQP依赖
<!--AMQP依赖,包含RabbitMQ-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
- publisher修改配置文件,并向Simple.queue队列中发送消息
spring:
rabbitmq:
host: 192.168.150.101 # rabbitMQ的ip地址
port: 5672 # 端口
username: itcast
password: 123321
virtual-host: /
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void testSendMessage2SimpleQueue() {
String queueName = "simple.queue";
String message = "hello, spring amqp!";
rabbitTemplate.convertAndSend(queueName, message);
}
- consumer修改配置文件,并绑定simple.queue队列消费消息
spring:
rabbitmq:
host: 192.168.150.101 # rabbitMQ的ip地址
port: 5672 # 端口
username: itcast
password: 123321
virtual-host: /
//定义的类记得加注解
@Component
public class SpringRabbitListener {
//方法需要加RabbitListener注解
@RabbitListener(queues = "simple.queue")
public void listenSimpleQueue(String msg) {
System.out.println("消费者接收到simple.queue的消息:【" + msg + "】");
}
}
- 注意:消息一旦消费就会从队列中删除,RabbitMQ没有消息回溯功能
九、Work Queue
- 多个消费者绑定到一个队列,同一条消息只会被一个消费者处理
- 结构如下
- 使用如下:
//Publisher
public void testSendMessage2WorkQueue() throws InterruptedException {
String queueName = "simple.queue";
String message = "hello, message__";
for (int i = 1; i <= 50; i++) {
rabbitTemplate.convertAndSend(queueName, message + i);
Thread.sleep(20);
}
}
//Consumer
@RabbitListener(queues = "simple.queue")
public void listenWorkQueue1(String msg) throws InterruptedException {
System.out.println("消费者1接收到消息:【" + msg + "】" + LocalTime.now());
Thread.sleep(20);
}
@RabbitListener(queues = "simple.queue")
public void listenWorkQueue2(String msg) throws InterruptedException {
System.err.println("消费者2........接收到消息:【" + msg + "】" + LocalTime.now());
Thread.sleep(200);
}
- 如果没有设置消息预取,消息就会被两个消费者平分;可以通过设置preFetch控制预取消息的上限
spring:
rabbitmq:
host: 192.168.150.101 # rabbitMQ的ip地址
port: 5672 # 端口
username: itcast
password: 123321
virtual-host: /
listener:
simple:
prefetch: 1
十、发布订阅模式
- 发布订阅模式允许将同一消息发送给多个消费者
- 注意:exchange(交换机)负责消息路由,而不是存储,路由失败则消息丢失
- 结构如下
十一、Fanout Exchange
- Fanout Exchange会将接收到的消息路由到每一个跟其绑定的queue
- 使用案例:
Publisher
//Publisher
public void testSendFanoutExchange() {
// 交换机名称
String exchangeName = "itcast.fanout";
// 消息
String message = "hello, every one!";
// 发送消息
rabbitTemplate.convertAndSend(exchangeName, "", message);
}
消费者可以用两种办法
// 其一:设置配置类
@Configuration
public class FanoutConfig {
@Bean
public FanoutExchange fanoutExchange(){
return new FanoutExchange("itcast.fanout");
}
// fanout.queue1
@Bean
public Queue fanoutQueue1(){
return new Queue("fanout.queue1");
}
// 绑定队列1到交换机
@Bean
public Binding fanoutBinding1(Queue fanoutQueue1, FanoutExchange fanoutExchange){
return BindingBuilder
.bind(fanoutQueue1)
.to(fanoutExchange);
}
// fanout.queue2
@Bean
public Queue fanoutQueue2(){
return new Queue("fanout.queue2");
}
// 绑定队列2到交换机
@Bean
public Binding fanoutBinding2(Queue fanoutQueue2, FanoutExchange fanoutExchange){
return BindingBuilder
.bind(fanoutQueue2)
.to(fanoutExchange);
}
@Bean
public Queue objectQueue(){
return new Queue("object.queue");
}
}
//其二:直接监听
@RabbitListener(queues = "fanout.queue1")
public void listenFanoutQueue1(String msg) {
System.out.println("消费者接收到fanout.queue1的消息:【" + msg + "】");
}
@RabbitListener(queues = "fanout.queue2")
public void listenFanoutQueue2(String msg) {
System.out.println("消费者接收到fanout.queue2的消息:【" + msg + "】");
}
十二、DirectExchange
- 路由模式,该交换器会将消息根据路由规则发送到指定的Queue中
- 每一个Queue都与Exchange设置一个BindingKey,以便匹配
- 使用案例如下:
Publisher
public void testSendDirectExchange() {
// 交换机名称
String exchangeName = "itcast.direct";
// 消息
String message = "hello, red!";
// 发送消息,red就是指定的BindingKey,可以改成其他的
rabbitTemplate.convertAndSend(exchangeName, "red", message);
}
Consumer
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "direct.queue1"),
exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
key = {"red", "blue"}
))
public void listenDirectQueue1(String msg){
System.out.println("消费者接收到direct.queue1的消息:【" + msg + "】");
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "direct.queue2"),
exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
key = {"red", "yellow"}
))
public void listenDirectQueue2(String msg){
System.out.println("消费者接收到direct.queue2的消息:【" + msg + "】");
}
十三、TopicExchange
- TopicExchange与Directchange类似,区别在于routinKey必须是多个单词的列表,并且以 . 分割
- 使用案例如下:
Publisher
public void testSendTopicExchange() {
// 交换机名称
String exchangeName = "itcast.topic";
// 消息
String message = "今天天气不错,我的心情好极了!";
// 发送消息
rabbitTemplate.convertAndSend(exchangeName, "china.weather", message);
}
Consumer
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "topic.queue1"),
exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),
key = "china.#"
))
public void listenTopicQueue1(String msg){
System.out.println("消费者接收到topic.queue1的消息:【" + msg + "】");
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "topic.queue2"),
exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),
key = "#.news"
))
public void listenTopicQueue2(String msg){
System.out.println("消费者接收到topic.queue2的消息:【" + msg + "】");
}
十四、消息转换器
- SpinrgAMQP发送和接收消息不仅仅是字符串,还可以是其他对象,发送过程会将对象序列化后进行发送
- 可以将发送的消息进行JSON方式序列化,消息会更简短易读
- 使用步骤:
在publisher和consumer服务中都引入依赖
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
</dependency>
在publisher和consumer服务的启动程序中声明MessageConverter,声明后底层就会自动将消息进行JSON序列化
@Bean
public MessageConverter messageConverter(){
return new Jackson2JsonMessageConverter();
}
然后consumer正常接受消息文章来源:https://www.toymoban.com/news/detail-479317.html
- SpringAMQP中消息的序列化和反序列化是如何实现的?
- 利用MessageConverter实现的,默认是JDK的序列化
- 注意发送方和接收方必须使用相同的MessageConverter
十五、消息堆积问题
- 当生产者发送消费的速度超过了消费者处理消息的速度,就会导致队列中的消息堆积,知道队列存储达到上限
- 解决有三种思路:
- 增加更多消费者,提高消费速度
- 在消费者内开启线程池加快消息处理速度
- 扩大队列容积,提高堆积上限
十六、惰性队列
- 从RabbitMQ的3.6.0版本后就有了
- 特征如下:
- 接收到消息后直接存入磁盘而非内存
- 消费者要消费消息时才会从磁盘中读取并加载到内存
- 支持数百万条的消息存储
- 优点在于性能稳定,消息上限高
- 缺点在于消息时效性降低,受限于磁盘IO
- 开启惰性队列有两种方式
- @ Bean的方式
- 注解方式:
- @ Bean的方式
十七、消息应答
- 消息应答就是:消费者在接收到消息并且处理该消息之后,告诉RabbitMQ它已经处理了,RabbitMQ可以把该消息删除了
- 自动应答:即消息发送后立即被认为已经传送成功,仅适用于消费者可以高效处理这些信息的情况下使用
- 手动应答:使用方法通知RabbitMQ该消息成功处理
- Channel.basicAck(用于肯定确认):成功处理该消息,可以丢弃了
- Channel.basicNack(用于否定确认):不处理该消息直接拒绝,可以丢弃了
- Channel.basicReject(用于否定确认):和basicNack相比少了一个批处理的参数
- 消息自动重新入队:若消费者由于某些原因失去连接,导致消息未发送ACK确认,RabbitMQ将了解到消息未完全处理,并将其重新排队,若此时存在其他消费者,将会重新分发给另一个消费者
十八、使用消息手动应答
- 默认消息采用的是自动应答,所以需要将其关闭
//basicConsume的第二个参数就是关闭自动应答
channel.basicConsume(TASK_QUEUE_NAME,false,deliverCallback,consumerTag->());
- 手动应答
//consumer的核心代码
Channel channel=RabbitUtils.getChannel();
channel.queueDeclare(TASK_QUEUE_NAME,false,false,false,null);
DeliverCallback deliverCallback=(consumerTag,delivery)->{
String message=new String(delivery.getBody(),"UTF-8");
//第一个参数指消息标记Tag
//第二个参数指是否批量应答消息,false指只应答当前传递的消息
channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
}
//basicConsume的第二个参数就是关闭自动应答
channel.basicConsume(TASK_QUEUE_NAME,false,deliverCallback,consumerTag->());
十九、RabbitMQ持久化
- 队列持久化:一般的队列在RabbitMQ重启后就会被删除,若想重启后队列依然存在,就需要在声明队列时把durable参数设置为持久化
//消费者将通道和队列绑定时将第二个参数设置为true
//参数依次为Queue名称、是否持久化、是否排外(加锁)、是否自动删除该队列、设置队列的其他一些参数
channel.queueDeclare(ACK_QUEUE_NAME,true,false,false,null);
- 注意:若想将一个正在运行的未持久化队列修改为持久化,需要将该队列先删除掉,然后重新创建一个持久化队列,否则会报错
- 消息持久化:即使设置了队列持久化,但当RabbitMQ重启时,里面的消息依然会丢失,因此需要将消息也设置为持久化才不会消息
- 原理就是高速RabbitMQ将消息保存到磁盘中,而不是内存中,代码实现如下
//生产者添加MessageProperties.PERSISTENT_TEXT_PLAIN参数
//参数依次为Exchange名称、Queue名称(或者说路由地址)、消息的配置属性(通常为null)、消息实体
channel.basicPublish("",TASK_QUEUE_NAME,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes("UTF-8"));
二十、发布确认
- 当信道被设置为confirm模式时,所有在该信道上发布的消息都将会被指派一个唯一的ID(从1开始),一旦消息成功投递到所有匹配的队列后,broker会发送一个确认给生产者(包含消息的唯一ID),这就是发布确认
- 开启发布确认的方法
//在创建channel后,调用confirmSelect方法即可
Channel channel=connection.createChannel();
channel.confirmSelect();
- 单个确认发布:即发布一个消息后只有被确认发布后才会发第二个消息
//核心代码
//即调用waitForConfirms等待发布的确认
for(...){
channel.basicPublish("",queueName,null,message.getBytes());
boolean flag=channel.waitForConfirms();
if(flag){
//TODO 其他业务
}
}
- 批量发布确认:等发布一批消息后再确认,其实就是加个条件判断,不要确认地这么频繁
//核心代码
for(...){
channel.basicPublish("",queueName,null,message.getBytes());
count++;
if(count%10==0){
channel.waitForConfirms();
//TODO 其他业务
}
}
- 异步确认发布:通过中间件的函数回调来判断是否投递成功,发布者只管发布
//发布者模块的核心代码
//消息确认成功 回调函数
//两个参数分别表示消息的标记和是否为批量确认
ConfirmCallback ackCallback=(deliveryTag,multiple)->{
System.out.println("确认的消息:"+deliveryTag);
};
//消息确认失败 回调函数
ConfirmCallback nackCallback=(deliveryTag,multiple)->{
System.out.println("未确认的消息:"+deliveryTag);
//TODO 其他业务
};
//准备消息的监听器,监听哪些消息成功或失败
channel.addConfirmListener(ackCallback,nackCallback);
//批量发送消息
for(...){
channel.basicPublish("",queueName,null,message.getBytes());
}
二十一、与Rabbit MQ相关的linux常用命令文章来源地址https://www.toymoban.com/news/detail-479317.html
- systemctl start rabbitmq-server.service
- systemctl status rabbitmq-server.service
- rabbitmqctl add_user admin admin
- rabbitmqctl set_permissions -p / admin “." ".” “.*”
- rabbitmqctl set_user_tag admin administrator
- rabbitmqctl set_user_tags admin administrator
- 开启管理页面:rabbitmq-plugins enable rabbitmq_management
- 关闭管理页面:rabbitmq-plugins disable rabbitmq_management
参考
- B站黑马2021SpringCloud课程:https://www.bilibili.com/video/BV1LQ4y127n4
- B站尚硅谷RabbitMQ教程:https://www.bilibili.com/video/BV1cb4y1o7zz
到了这里,关于SpringCloud学习笔记(四)RabbitMQ的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!