网站设计与建设,苏州智能网站建设,遂宁北京网站建设,动漫制作专业大一需不需要买电脑服务异步通信RabbitMQ-高级篇服务异步通信RabbitMQ-高级篇1.消息可靠性1.1.生产者消息确认1.1.1.修改配置1.1.2.定义Return回调1.1.3.定义ConfirmCallback1.2.消息持久化1.2.1.交换机持久化1.2.2.队列持久化1.2.3.消息持久化1.3.消费者消息确认1.3.1.演示none模式1.3.2.演示aut…
服务异步通信RabbitMQ-高级篇服务异步通信RabbitMQ-高级篇1.消息可靠性1.1.生产者消息确认1.1.1.修改配置1.1.2.定义Return回调1.1.3.定义ConfirmCallback1.2.消息持久化1.2.1.交换机持久化1.2.2.队列持久化1.2.3.消息持久化1.3.消费者消息确认1.3.1.演示none模式1.3.2.演示auto模式1.4.消费失败重试机制1.4.1.本地重试1.4.2.失败策略1.5.总结2.死信交换机2.1.初识死信交换机2.1.1.什么是死信交换机2.1.2.利用死信交换机接收死信拓展2.1.3.总结2.2.TTL2.2.1.接收超时死信的死信交换机2.2.2.声明一个队列并且指定TTL2.2.3.发送消息时设定TTL2.2.4.总结2.3.延迟队列2.3.1.安装DelayExchange插件2.3.2.DelayExchange原理2.3.3.使用DelayExchange1声明DelayExchange交换机2发送消息2.3.4.总结3.惰性队列3.1.消息堆积问题3.2.惰性队列3.2.1.基于命令行设置lazy-queue3.2.2.基于Bean声明lazy-queue3.2.3.基于RabbitListener声明LazyQueue3.3.总结4.MQ集群4.1.集群分类4.2.普通集群4.2.1.集群结构和特征4.2.2.部署4.3.镜像集群4.3.1.集群结构和特征4.3.2.部署4.4.仲裁队列4.4.1.集群特征4.4.2.部署4.4.3.Java代码创建仲裁队列4.4.4.SpringAMQP连接MQ集群服务异步通信RabbitMQ-高级篇
消息队列在使用过程中面临着很多实际问题需要思考 1.消息可靠性
消息从发送到消费者接收会经理多个过程 其中的每一步都可能导致消息丢失常见的丢失原因包括
发送时丢失 生产者发送的消息未送达exchange消息到达exchange后未到达queue MQ宕机queue将消息丢失consumer接收到消息后未消费就宕机
针对这些问题RabbitMQ分别给出了解决方案
生产者确认机制mq持久化消费者确认机制失败重试机制
1.1.生产者消息确认
RabbitMQ提供了publisher confirm机制来避免消息发送到MQ过程中丢失。这种机制必须给每个消息指定一个唯一ID。消息发送到MQ以后会返回一个结果给发送者表示消息是否处理成功。
返回结果有两种方式
publisher-confirm发送者确认 消息成功投递到交换机返回ack消息未投递到交换机返回nack publisher-return发送者回执 消息投递到交换机了但是没有路由到队列。返回ACK及路由失败原因。 注意 1.1.1.修改配置
首先修改publisher服务中的application.yml文件添加下面的内容
spring:rabbitmq:publisher-confirm-type: correlatedpublisher-returns: truetemplate:mandatory: true
说明
publish-confirm-type开启publisher-confirm这里支持两种类型 simple同步等待confirm结果直到超时correlated异步回调定义ConfirmCallbackMQ返回结果时会回调这个ConfirmCallback publish-returns开启publish-return功能同样是基于callback机制不过是定义ReturnCallbacktemplate.mandatory定义消息路由失败时的策略。true则调用ReturnCallbackfalse则直接丢弃消息
1.1.2.定义Return回调
每个RabbitTemplate只能配置一个ReturnCallback因此需要在项目加载时配置
修改publisher服务添加一个
package cn.itcast.mq.config;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Configuration;Slf4j
Configuration
public class CommonConfig implements ApplicationContextAware {Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {// 获取RabbitTemplateRabbitTemplate rabbitTemplate applicationContext.getBean(RabbitTemplate.class);// 设置ReturnCallbackrabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) - {// 投递失败记录日志log.info(消息发送失败应答码{}原因{}交换机{}路由键{},消息{},replyCode, replyText, exchange, routingKey, message.toString());// 如果有业务需要可以重发消息});}
}1.1.3.定义ConfirmCallback
ConfirmCallback可以在发送消息时指定因为每个业务处理confirm成功或失败的逻辑不一定相同。
在publisher服务的cn.itcast.mq.spring.SpringAmqpTest类中定义一个单元测试方法
public void testSendMessage2SimpleQueue() throws InterruptedException {// 1.消息体String message hello, spring amqp!;// 2.全局唯一的消息ID需要封装到CorrelationData中CorrelationData correlationData new CorrelationData(UUID.randomUUID().toString());// 3.添加callbackcorrelationData.getFuture().addCallback(result - {if(result.isAck()){// 3.1.ack消息成功log.debug(消息发送成功, ID:{}, correlationData.getId());}else{// 3.2.nack消息失败log.error(消息发送失败, ID:{}, 原因{},correlationData.getId(), result.getReason());}},ex - log.error(消息发送异常, ID:{}, 原因{},correlationData.getId(),ex.getMessage()));// 4.发送消息rabbitTemplate.convertAndSend(task.direct, task, message, correlationData);// 休眠一会儿等待ack回执Thread.sleep(2000);
}1.2.消息持久化
生产者确认可以确保消息投递到RabbitMQ的队列中但是消息发送到RabbitMQ以后如果突然宕机也可能导致消息丢失。
要想确保消息在RabbitMQ中安全保存必须开启消息持久化机制。
交换机持久化队列持久化消息持久化
1.2.1.交换机持久化
RabbitMQ中交换机默认是非持久化的mq重启后就丢失。
SpringAMQP中可以通过代码指定交换机持久化
Bean
public DirectExchange simpleExchange(){// 三个参数交换机名称、是否持久化、当没有queue与其绑定时是否自动删除return new DirectExchange(simple.direct, true, false);
}事实上默认情况下由SpringAMQP声明的交换机都是持久化的。
可以在RabbitMQ控制台看到持久化的交换机都会带上D的标示 1.2.2.队列持久化
RabbitMQ中队列默认是非持久化的mq重启后就丢失。
SpringAMQP中可以通过代码指定交换机持久化
Bean
public Queue simpleQueue(){// 使用QueueBuilder构建队列durable就是持久化的return QueueBuilder.durable(simple.queue).build();
}事实上默认情况下由SpringAMQP声明的队列都是持久化的。
可以在RabbitMQ控制台看到持久化的队列都会带上D的标示 1.2.3.消息持久化
利用SpringAMQP发送消息时可以设置消息的属性MessageProperties指定delivery-mode
1非持久化2持久化
用java代码指定 默认情况下SpringAMQP发出的任何消息都是持久化的不用特意指定。
1.3.消费者消息确认
RabbitMQ是阅后即焚机制RabbitMQ确认消息被消费者消费后会立刻删除。
而RabbitMQ是通过消费者回执来确认消费者是否成功处理消息的消费者获取消息后应该向RabbitMQ发送ACK回执表明自己已经处理消息。
设想这样的场景
1RabbitMQ投递消息给消费者2消费者获取消息后返回ACK给RabbitMQ3RabbitMQ删除消息4消费者宕机消息尚未处理
这样消息就丢失了。因此消费者返回ACK的时机非常重要。
而SpringAMQP则允许配置三种确认模式
•manual手动ack需要在业务代码结束后调用api发送ack。
•auto自动ack由spring监测listener代码是否出现异常没有异常则返回ack抛出异常则返回nack
•none关闭ackMQ假定消费者获取消息后会成功处理因此消息投递后立即被删除
由此可知
none模式下消息投递是不可靠的可能丢失auto模式类似事务机制出现异常时返回nack消息回滚到mq没有异常返回ackmanual自己根据业务情况判断什么时候该ack
一般我们都是使用默认的auto即可。
1.3.1.演示none模式
修改consumer服务的application.yml文件添加下面内容
spring:rabbitmq:listener:simple:acknowledge-mode: none # 关闭ack修改consumer服务的SpringRabbitListener类中的方法模拟一个消息处理异常
RabbitListener(queues simple.queue)
public void listenSimpleQueue(String msg) {log.info(消费者接收到simple.queue的消息【{}】, msg);// 模拟异常System.out.println(1 / 0);log.debug(消息处理完成);
}测试可以发现当消息处理抛异常时消息依然被RabbitMQ删除了。
1.3.2.演示auto模式
再次把确认机制修改为auto:
spring:rabbitmq:listener:simple:acknowledge-mode: auto # 关闭ack在异常位置打断点再次发送消息程序卡在断点时可以发现此时消息状态为unack未确定状态 抛出异常后因为Spring会自动返回nack所以消息恢复至Ready状态并且没有被RabbitMQ删除 1.4.消费失败重试机制
当消费者出现异常后消息会不断requeue重入队到队列再重新发送给消费者然后再次异常再次requeue无限循环导致mq的消息处理飙升带来不必要的压力 怎么办呢
1.4.1.本地重试
我们可以利用Spring的retry机制在消费者出现异常时利用本地重试而不是无限制的requeue到mq队列。
修改consumer服务的application.yml文件添加内容
spring:rabbitmq:listener:simple:retry:enabled: true # 开启消费者失败重试initial-interval: 1000 # 初识的失败等待时长为1秒multiplier: 1 # 失败的等待时长倍数下次等待时长 multiplier * last-intervalmax-attempts: 3 # 最大重试次数stateless: true # true无状态false有状态。如果业务中包含事务这里改为false
重启consumer服务重复之前的测试。可以发现
在重试3次后SpringAMQP会抛出异常AmqpRejectAndDontRequeueException说明本地重试触发了查看RabbitMQ控制台发现消息被删除了说明最后SpringAMQP返回的是ackmq删除消息了
结论
开启本地重试时消息处理过程中抛出异常不会requeue到队列而是在消费者本地重试重试达到最大次数后Spring会返回ack消息会被丢弃
1.4.2.失败策略
在之前的测试中达到最大重试次数后消息会被丢弃这是由Spring内部机制决定的。
在开启重试模式后重试次数耗尽如果消息依然失败则需要有MessageRecovery接口来处理它包含三种不同的实现
RejectAndDontRequeueRecoverer重试耗尽后直接reject丢弃消息。默认就是这种方式ImmediateRequeueMessageRecoverer重试耗尽后返回nack消息重新入队RepublishMessageRecoverer重试耗尽后将失败消息投递到指定的交换机
比较优雅的一种处理方案是RepublishMessageRecoverer失败后将消息投递到一个指定的专门存放异常消息的队列后续由人工集中处理。
1在consumer服务中定义处理失败消息的交换机和队列
Bean
public DirectExchange errorMessageExchange(){return new DirectExchange(error.direct);
}
Bean
public Queue errorQueue(){return new Queue(error.queue, true);
}
Bean
public Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with(error);
}
2定义一个RepublishMessageRecoverer关联队列和交换机
Bean
public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){return new RepublishMessageRecoverer(rabbitTemplate, error.direct, error);
}
完整代码
package cn.itcast.mq.config;import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
import org.springframework.context.annotation.Bean;Configuration
public class ErrorMessageConfig {Beanpublic DirectExchange errorMessageExchange(){return new DirectExchange(error.direct);}Beanpublic Queue errorQueue(){return new Queue(error.queue, true);}Beanpublic Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with(error);}Beanpublic MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){return new RepublishMessageRecoverer(rabbitTemplate, error.direct, error);}
}
1.5.总结
如何确保RabbitMQ消息的可靠性
开启生产者确认机制确保生产者的消息能到达队列开启持久化功能确保消息未消费前在队列中不会丢失开启消费者确认机制为auto由spring确认消息处理成功后完成ack开启消费者失败重试机制并设置MessageRecoverer多次重试失败后将消息投递到异常交换机交由人工处理
2.死信交换机
2.1.初识死信交换机
2.1.1.什么是死信交换机
什么是死信
当一个队列中的消息满足下列情况之一时可以成为死信dead letter
消费者使用basic.reject或 basic.nack声明消费失败并且消息的requeue参数设置为false消息是一个过期消息超时无人消费要投递的队列消息满了无法投递
如果这个包含死信的队列配置了dead-letter-exchange属性指定了一个交换机那么队列中的死信就会投递到这个交换机中而这个交换机称为死信交换机Dead Letter Exchange检查DLX。
如图一个消息被消费者拒绝了变成了死信 因为simple.queue绑定了死信交换机 dl.direct因此死信会投递给这个交换机 如果这个死信交换机也绑定了一个队列则消息最终会进入这个存放死信的队列 另外队列将死信投递给死信交换机时必须知道两个信息
死信交换机名称死信交换机与死信队列绑定的RoutingKey
这样才能确保投递的消息能到达死信交换机并且正确的路由到死信队列。 2.1.2.利用死信交换机接收死信拓展
在失败重试策略中默认的RejectAndDontRequeueRecoverer会在本地重试次数耗尽后发送reject给RabbitMQ消息变成死信被丢弃。
我们可以给simple.queue添加一个死信交换机给死信交换机绑定一个队列。这样消息变成死信后也不会丢弃而是最终投递到死信交换机路由到与死信交换机绑定的队列。 我们在consumer服务中定义一组死信交换机、死信队列
// 声明普通的 simple.queue队列并且为其指定死信交换机dl.direct
Bean
public Queue simpleQueue2(){return QueueBuilder.durable(simple.queue) // 指定队列名称并持久化.deadLetterExchange(dl.direct) // 指定死信交换机.build();
}
// 声明死信交换机 dl.direct
Bean
public DirectExchange dlExchange(){return new DirectExchange(dl.direct, true, false);
}
// 声明存储死信的队列 dl.queue
Bean
public Queue dlQueue(){return new Queue(dl.queue, true);
}
// 将死信队列 与 死信交换机绑定
Bean
public Binding dlBinding(){return BindingBuilder.bind(dlQueue()).to(dlExchange()).with(simple);
}
2.1.3.总结
什么样的消息会成为死信
消息被消费者reject或者返回nack消息超时未消费队列满了
死信交换机的使用场景是什么
如果队列绑定了死信交换机死信会投递到死信交换机可以利用死信交换机收集所有消费者处理失败的消息死信交由人工处理进一步提高消息队列的可靠性。
2.2.TTL
一个队列中的消息如果超时未消费则会变为死信超时分为两种情况
消息所在的队列设置了超时时间消息本身设置了超时时间 2.2.1.接收超时死信的死信交换机
在consumer服务的SpringRabbitListener中定义一个新的消费者并且声明 死信交换机、死信队列
RabbitListener(bindings QueueBinding(value Queue(name dl.ttl.queue, durable true),exchange Exchange(name dl.ttl.direct),key ttl
))
public void listenDlQueue(String msg){log.info(接收到 dl.ttl.queue的延迟消息{}, msg);
}
2.2.2.声明一个队列并且指定TTL
要给队列设置超时时间需要在声明队列时配置x-message-ttl属性
Bean
public Queue ttlQueue(){return QueueBuilder.durable(ttl.queue) // 指定队列名称并持久化.ttl(10000) // 设置队列的超时时间10秒.deadLetterExchange(dl.ttl.direct) // 指定死信交换机.build();
}
注意这个队列设定了死信交换机为dl.ttl.direct
声明交换机将ttl与交换机绑定
Bean
public DirectExchange ttlExchange(){return new DirectExchange(ttl.direct);
}
Bean
public Binding ttlBinding(){return BindingBuilder.bind(ttlQueue()).to(ttlExchange()).with(ttl);
}
发送消息但是不要指定TTL
Test
public void testTTLQueue() {// 创建消息String message hello, ttl queue;// 消息ID需要封装到CorrelationData中CorrelationData correlationData new CorrelationData(UUID.randomUUID().toString());// 发送消息rabbitTemplate.convertAndSend(ttl.direct, ttl, message, correlationData);// 记录日志log.debug(发送消息成功);
}
发送消息的日志 查看下接收消息的日志 因为队列的TTL值是10000ms也就是10秒。可以看到消息发送与接收之间的时差刚好是10秒。
2.2.3.发送消息时设定TTL
在发送消息时也可以指定TTL
Test
public void testTTLMsg() {// 创建消息Message message MessageBuilder.withBody(hello, ttl message.getBytes(StandardCharsets.UTF_8)).setExpiration(5000).build();// 消息ID需要封装到CorrelationData中CorrelationData correlationData new CorrelationData(UUID.randomUUID().toString());// 发送消息rabbitTemplate.convertAndSend(ttl.direct, ttl, message, correlationData);log.debug(发送消息成功);
}
查看发送消息日志 接收消息日志 这次发送与接收的延迟只有5秒。说明当队列、消息都设置了TTL时任意一个到期就会成为死信。
2.2.4.总结
消息超时的两种方式是
给队列设置ttl属性进入队列后超过ttl时间的消息变为死信给消息设置ttl属性队列接收到消息超过ttl时间后变为死信
如何实现发送一个消息20秒后消费者才收到消息
给消息的目标队列指定死信交换机将消费者监听的队列绑定到死信交换机发送消息时给消息设置超时时间为20秒
2.3.延迟队列
利用TTL结合死信交换机我们实现了消息发出后消费者延迟收到消息的效果。这种消息模式就称为延迟队列Delay Queue模式。
延迟队列的使用场景包括
延迟发送短信用户下单如果用户在15 分钟内未支付则自动取消预约工作会议20分钟后自动通知所有参会人员
因为延迟队列的需求非常多所以RabbitMQ的官方也推出了一个插件原生支持延迟队列效果。
这个插件就是DelayExchange插件。参考RabbitMQ的插件列表页面https://www.rabbitmq.com/community-plugins.html 使用方式可以参考官网地址https://blog.rabbitmq.com/posts/2015/04/scheduling-messages-with-rabbitmq
2.3.1.安装DelayExchange插件
参考资料《RabbitMQ部署指南.md》
2.3.2.DelayExchange原理
DelayExchange需要将一个交换机声明为delayed类型。当我们发送消息到delayExchange时流程如下
接收消息判断消息是否具备x-delay属性如果有x-delay属性说明是延迟消息持久化到硬盘读取x-delay值作为延迟时间返回routing not found结果给消息发送者x-delay时间到期后重新投递消息到指定队列
2.3.3.使用DelayExchange
插件的使用也非常简单声明一个交换机交换机的类型可以是任意类型只需要设定delayed属性为true即可然后声明队列与其绑定即可。
1声明DelayExchange交换机
基于注解方式推荐 也可以基于Bean的方式 2发送消息
发送消息时一定要携带x-delay属性指定延迟的时间 2.3.4.总结
延迟队列插件的使用步骤包括哪些
•声明一个交换机添加delayed属性为true
•发送消息时添加x-delay头值为超时时间
3.惰性队列
3.1.消息堆积问题
当生产者发送消息的速度超过了消费者处理消息的速度就会导致队列中的消息堆积直到队列存储消息达到上限。之后发送的消息就会成为死信可能会被丢弃这就是消息堆积问题。 解决消息堆积有两种思路
增加更多消费者提高消费速度。也就是我们之前说的work queue模式扩大队列容积提高堆积上限
要提升队列容积把消息保存在内存中显然是不行的。
3.2.惰性队列
从RabbitMQ的3.6.0版本开始就增加了Lazy Queues的概念也就是惰性队列。惰性队列的特征如下
接收到消息后直接存入磁盘而非内存消费者要消费消息时才会从磁盘中读取并加载到内存支持数百万条的消息存储
3.2.1.基于命令行设置lazy-queue
而要设置一个队列为惰性队列只需要在声明队列时指定x-queue-mode属性为lazy即可。可以通过命令行将一个运行中的队列修改为惰性队列
rabbitmqctl set_policy Lazy ^lazy-queue$ {queue-mode:lazy} --apply-to queues
命令解读
rabbitmqctl RabbitMQ的命令行工具set_policy 添加一个策略Lazy 策略名称可以自定义^lazy-queue$ 用正则表达式匹配队列的名字{queue-mode:lazy} 设置队列模式为lazy模式--apply-to queues 策略的作用对象是所有的队列
3.2.2.基于Bean声明lazy-queue 3.2.3.基于RabbitListener声明LazyQueue 3.3.总结
消息堆积问题的解决方案
队列上绑定多个消费者提高消费速度使用惰性队列可以再mq中保存更多消息
惰性队列的优点有哪些
基于磁盘存储消息上限高没有间歇性的page-out性能比较稳定
惰性队列的缺点有哪些
基于磁盘存储消息时效性会降低性能受限于磁盘的IO
4.MQ集群
4.1.集群分类
RabbitMQ的是基于Erlang语言编写而Erlang又是一个面向并发的语言天然支持集群模式。RabbitMQ的集群有两种模式
•普通集群是一种分布式集群将队列分散到集群的各个节点从而提高整个集群的并发能力。
•镜像集群是一种主从集群普通集群的基础上添加了主从备份功能提高集群的数据可用性。
镜像集群虽然支持主从但主从同步并不是强一致的某些情况下可能有数据丢失的风险。因此在RabbitMQ的3.8版本以后推出了新的功能仲裁队列来代替镜像集群底层采用Raft协议确保主从的数据一致性。
4.2.普通集群
4.2.1.集群结构和特征
普通集群或者叫标准集群classic cluster具备下列特征
会在集群的各个节点间共享部分数据包括交换机、队列元信息。不包含队列中的消息。当访问集群某节点时如果队列不在该节点会从数据所在节点传递到当前节点并返回队列所在节点宕机队列中的消息就会丢失
结构如图 4.2.2.部署
参考资料《RabbitMQ部署指南.md》
4.3.镜像集群
4.3.1.集群结构和特征
镜像集群本质是主从模式具备下面的特征
交换机、队列、队列中的消息会在各个mq的镜像节点之间同步备份。创建队列的节点被称为该队列的主节点备份到的其它节点叫做该队列的镜像节点。一个队列的主节点可能是另一个队列的镜像节点所有操作都是主节点完成然后同步给镜像节点主宕机后镜像节点会替代成新的主
结构如图 4.3.2.部署
参考资料《RabbitMQ部署指南.md》
4.4.仲裁队列
4.4.1.集群特征
仲裁队列仲裁队列是3.8版本以后才有的新功能用来替代镜像队列具备下列特征
与镜像队列一样都是主从模式支持主从数据同步使用非常简单没有复杂的配置主从同步基于Raft协议强一致
4.4.2.部署
参考资料《RabbitMQ部署指南.md》
4.4.3.Java代码创建仲裁队列
Bean
public Queue quorumQueue() {return QueueBuilder.durable(quorum.queue) // 持久化.quorum() // 仲裁队列.build();
}
4.4.4.SpringAMQP连接MQ集群
注意这里用address来代替host、port方式
spring:rabbitmq:addresses: 192.168.150.105:8071, 192.168.150.105:8072, 192.168.150.105:8073username: itcastpassword: 123321virtual-host: /
