山东省乡镇网站建设,上海网站优化推广,网站首页代码,如何做闲置物品交换的网站SpringCloud微服务技术栈.黑马跟学 十二今日目标服务异步通信-高级篇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…
SpringCloud微服务技术栈.黑马跟学 十二今日目标服务异步通信-高级篇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插件1.安装DelayExchange插件2.下载插件3.上传插件4.安装插件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.部署1.集群分类2.获取cookie3.准备集群配置4.启动集群5.测试6.数据共享测试7.可用性测试4.3.镜像集群4.3.1.集群结构和特征4.3.2.部署1.镜像模式的特征2.镜像模式的配置3.exactly模式4.模式5.nodes模式6.测试7.测试数据共享8.测试高可用5.仲裁队列4.4.仲裁队列4.4.1.集群特征4.4.2.部署1.添加仲裁队列2.测试3.集群扩容4.加入集群5.增加仲裁队列副本4.4.3.Java代码创建仲裁队列4.4.4.SpringAMQP连接MQ集群今日目标 服务异步通信-高级篇
消息队列在使用过程中面临着很多实际问题需要思考
1.消息可靠性
消息从发送到消费者接收会经理多个过程 其中的每一步都可能导致消息丢失常见的丢失原因包括
发送时丢失 生产者发送的消息未送达exchange消息到达exchange后未到达queue MQ宕机queue将消息丢失consumer接收到消息后未消费就宕机
针对这些问题RabbitMQ分别给出了解决方案
生产者确认机制mq持久化消费者确认机制失败重试机制
下面我们就通过案例来演示每一个步骤。 首先导入课前资料提供的demo工程 项目结构如下 用docker启动即可
docker start mq要创建一个队列起名simple.queue 然后在交换机中把amq.topic交换机和上面创建的队列simple.queue绑定我们手动配置 进入amq.topic交换机后绑定队列 绑定后如图
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);
}全部配置完后运行测试类SpringAmqpTest.java这说明消息发送成功 然后呢我们来一个消息发送失败的情况我们故意填错交换机的名字 调用后后台打印日志如下 然后我们尝试填错routingKey看一下 报错信息如下 之后我们恢复代码都保证正确即可
总结 SpringAMQP中处理消息确认的几种情况: ● publisher-comfirm:
消息成功发送到exchange,返回ack消息发送失败没有到达交换机返回nack消息发送过程中出现异常没有收到回执
● 消息成功发送到exchange, 但没有路由到queue,
调用ReturnCallback
1.2.消息持久化
生产者确认可以确保消息投递到RabbitMQ的队列中但是消息发送到RabbitMQ以后如果突然宕机也可能导致消息丢失。
要想确保消息在RabbitMQ中安全保存必须开启消息持久化机制。
交换机持久化队列持久化消息持久化
1.2.1.交换机持久化
RabbitMQ中交换机默认是非持久化的mq重启后就丢失。
我们通过命令 重启mq
docker restart mq然后查看队列、交换机的情况比如我们创建的是持久化队列 SpringAMQP中可以通过代码指定交换机持久化
Bean
public DirectExchange simpleExchange(){// 三个参数交换机名称、是否持久化、当没有queue与其绑定时是否自动删除return new DirectExchange(simple.direct, true, false);
}事实上默认情况下由SpringAMQP声明的交换机都是持久化的。
可以在RabbitMQ控制台看到持久化的交换机都会带上D的标示
1.2.2.队列持久化
RabbitMQ中队列默认是非持久化的mq重启后就丢失。 SpringAMQP中可以通过代码指定交换机持久化
我们可以先去mq图形化界面把simple.queue删除
Bean
public Queue simpleQueue(){// 使用QueueBuilder构建队列durable就是持久化的return QueueBuilder.durable(simple.queue).build();
}事实上默认情况下由SpringAMQP声明的队列都是持久化的。 可以在RabbitMQ控制台看到持久化的队列都会带上D的标示
这些做完后我们启动ConsumerApplication.java然后查看mq的图形化界面 交换机是持久的 队列是持久的
1.2.3.消息持久化
首先把consumer服务停了不要消费我们的消息 我们在mq的图形化界面点击simple.queue队列然后编辑消息点击发送 查看有1条消息 然后我们重启docker中的mq
docker restart mq然后再回来看mq的图形化界面发现队列还在但是消息没了
利用SpringAMQP发送消息时可以设置消息的属性MessageProperties指定delivery-mode
1非持久化2持久化
用java代码指定
默认情况下SpringAMQP发出的任何消息都是持久化的不用特意指定。 运行测试类SpringAmqpTest.java之后查看mq的图形化界面 查看一下具体消息 然后我们重启一下docker的mq容器
docker restart mq注意AMQP中创建的交换机、队列、消息默认都是持久的 交换机 队列
消息
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类中的方法模拟一个消息处理异常 修改SpringRabbitListener.java
RabbitListener(queues simple.queue)
public void listenSimpleQueue(String msg) {log.info(消费者接收到simple.queue的消息【{}】, msg);// 模拟异常System.out.println(1 / 0);log.debug(消息处理完成);
}测试可以发现当消息处理抛异常时消息依然被RabbitMQ删除了。 dubug启动Consumer 发现消息还没接收呢直接就没了 也就是说消费者虽然接收到了消息但是假如消费者还没有读取发生了报错或者宕机这个消息就会丢失
1.3.2.演示auto模式
再次把确认机制修改为auto:
spring:rabbitmq:listener:simple:acknowledge-mode: auto # 关闭ack我们去mq的图形化界面创建消息 发送后我们看到图形化界面中有1条消息 IDEA后台因为我们认为写了1/0的错误算数运算导致IDEA不停重发请求重试消息的推送这显然也不符合我们的要求
在异常位置打断点再次发送消息程序卡在断点时可以发现此时消息状态为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: 4 # 最大重试次数stateless: true # true无状态false有状态。如果业务中包含事务这里改为false修改SpringRabbitListener.java 修改为日志打印的形式 RabbitListener(queues simple.queue)public void listenSimpleQueue(String msg) {log.debug(消费者接收到simple.queue的消息【 msg 】);System.out.println(1 / 0);log.info(消费者处理消息成功!);}重启consumer服务重复之前的测试。可以发现
在重试4次后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);}
}以上配置完之后我们再重复步骤发送消息 发送后我们看到失败交换机有了 队列也有了 看一下IDEA的后台 看一下error.queue中的消息很清晰把错误栈都输出了
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插件
参考课前资料
1.安装DelayExchange插件
官方的安装指南地址为https://blog.rabbitmq.com/posts/2015/04/scheduling-messages-with-rabbitmq
上述文档是基于linux原生安装RabbitMQ然后安装插件。
因为我们之前是基于Docker安装RabbitMQ所以下面我们会讲解基于Docker来安装RabbitMQ插件。
2.下载插件
RabbitMQ有一个官方的插件社区地址为https://www.rabbitmq.com/community-plugins.html
其中包含各种各样的插件包括我们要使用的DelayExchange插件 大家可以去对应的GitHub页面下载3.8.9版本的插件地址为https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/tag/3.8.9这个对应RabbitMQ的3.8.5以上版本。
课前资料也提供了下载好的插件
3.上传插件
因为我们是基于Docker安装所以需要先查看RabbitMQ的插件目录对应的数据卷。如果不是基于Docker的同学请参考第一章部分重新创建Docker容器。
我们之前设定的RabbitMQ的数据卷名称为mq-plugins所以我们使用下面命令查看数据卷
docker volume inspect mq-plugins可以得到下面结果 接下来将插件上传到这个目录即可
4.安装插件
最后就是安装了需要进入MQ容器内部来执行安装。我的容器名为mq所以执行下面命令
docker exec -it mq bash执行时请将其中的 -it 后面的mq替换为你自己的容器名.
进入容器内部后执行下面命令开启插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange结果如下
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交换机
基于注解方式推荐 SpringRabbitListener.java
/*** 延迟交换机和队列** param message*/RabbitListener(bindings QueueBinding(value Queue(name delay.queue, durable true),exchange Exchange(name delay.direct, delayed true),key delay))public void listenDelayExchange(String message) {log.info(消费者接收到了delay.queue的消息 message);}也可以基于Bean的方式
2发送消息
发送消息时一定要携带x-delay属性指定延迟的时间 SpringAmqpTest.java /*** 发送延迟消息** throws InterruptedException*/Testpublic void testSendDealyMessage() throws InterruptedException {String routingKey delay;// 创建消息Message message MessageBuilder.withBody(hello, delay message !.getBytes(StandardCharsets.UTF_8)).setDeliveryMode(MessageDeliveryMode.PERSISTENT).setHeader(x-delay, 5000).build();// 准备CorrelationDataCorrelationData correlationData new CorrelationData(UUID.randomUUID().toString());rabbitTemplate.convertAndSend(delay.direct, routingKey, message, correlationData);}log.info(发送delay消息成功!);看一下mq的图形界面
运行测试类SpringAmqpTest.java发送消息虽然发送delay消息成功但是下面报了错误 查看consumer,5秒后接收消息成功
那这里为什么会报错呢这是因为delay的交换机是将消息持有了5秒后发送的这里其实不是报错而是消息暂存了5秒过了5秒才发送到队列那我们能不让它们报错吗当然可以我们继续修改。 根据receivedDelay是否有值判断是否重发有就不重发 修改publisher中的CommonConfig.java Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {// 获取RabbitTemplate对象RabbitTemplate rabbitTemplate applicationContext.getBean(RabbitTemplate.class);// 配置eturnCallbackrabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) - {// 判断是否是延迟消息if (message.getMessageProperties().getReceivedDelay() 0) {// 是一个延迟消息忽略报错信息return;}// 记录日志log.error(消息发送到队列失败响应码:{},失败原因:{},交换机:{},routingKey:{},消息:{},replyCode, replyText, exchange, routingKey, message);// 如果有失败的可以进行消息的重发});}再此通过测试类发送消息报错就没有了
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 新建类LazyConfig.java
Configuration
public class LazyConfig {/*** 惰性队列** return*/Beanpublic Queue lazyQueue() {return QueueBuilder.durable(lazy.queue).lazy().build();}/*** 普通队列** return*/Beanpublic Queue normalQueue() {return QueueBuilder.durable(normal.queue).build();}
}
启动Consumer服务查看mq图形界面 修改SpringAmqpTest.java /*** 测试惰性队列** throws InterruptedException*/Testpublic void testSendLazyQueue() throws InterruptedException {for (int i 0; i 1000000; i) {String routingKey lazy.queue;// 创建消息Message message MessageBuilder.withBody(hello, lazy queue.getBytes(StandardCharsets.UTF_8)).setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT).build();rabbitTemplate.convertAndSend(routingKey, message);//log.info(发送lazy队列消息成功!);}}/*** 测试惰性队列** throws InterruptedException*/Testpublic void testSendNormalQueue() throws InterruptedException {for (int i 0; i 1000000; i) {String routingKey normal.queue;// 创建消息Message message MessageBuilder.withBody(hello, normal queue.getBytes(StandardCharsets.UTF_8)).setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT).build();rabbitTemplate.convertAndSend(routingKey, message);//log.info(发送normal队列消息成功!);}}我们看到lazy queue中memory中没有直接写磁盘比较稳定 我们看normal 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》
集群部署 接下来我们看看如何安装RabbitMQ的集群。
1.集群分类
在RabbitMQ的官方文档中讲述了两种集群的配置方式
普通模式普通模式集群不进行数据同步每个MQ都有自己的队列、数据信息其它元数据信息如交换机等会同步。例如我们有2个MQmq1和mq2如果你的消息在mq1而你连接到了mq2那么mq2会去mq1拉取消息然后返回给你。如果mq1宕机消息就会丢失。镜像模式与普通模式不同队列会在各个mq的镜像节点之间同步因此你连接到任何一个镜像节点均可获取到消息。而且如果一个节点宕机并不会导致数据丢失。不过这种方式增加了数据同步的带宽消耗。
我们先来看普通模式集群我们的计划部署3节点的mq集群
主机名控制台端口amqp通信端口mq18081 — 156728071 — 5672mq28082 — 156728072 — 5672mq38083 — 156728073 — 5672
集群中的节点标示默认都是rabbit[hostname]因此以上三个节点的名称分别为
rabbitmq1rabbitmq2rabbitmq3
2.获取cookie
RabbitMQ底层依赖于Erlang而Erlang虚拟机就是一个面向分布式的语言默认就支持集群模式。集群模式中的每个RabbitMQ 节点使用 cookie 来确定它们是否被允许相互通信。
要使两个节点能够通信它们必须具有相同的共享秘密称为Erlang cookie。cookie 只是一串最多 255 个字符的字母数字字符。
每个集群节点必须具有相同的 cookie。实例之间也需要它来相互通信。
我们先在之前启动的mq容器中获取一个cookie值作为集群的cookie。执行下面的命令
docker exec -it mq cat /var/lib/rabbitmq/.erlang.cookie可以看到cookie值如下
FXZMCVGLBIXZCDEMMVZQ接下来停止并删除当前的mq容器我们重新搭建集群。
docker rm -f mq清理数据卷
docker volume prune3.准备集群配置
在/tmp目录新建一个配置文件 rabbitmq.conf
cd /tmp
# 创建文件
touch rabbitmq.conf文件内容如下
loopback_users.guest false
listeners.tcp.default 5672
cluster_formation.peer_discovery_backend rabbit_peer_discovery_classic_config
cluster_formation.classic_config.nodes.1 rabbitmq1
cluster_formation.classic_config.nodes.2 rabbitmq2
cluster_formation.classic_config.nodes.3 rabbitmq3再创建一个文件记录cookie
cd /tmp
# 创建cookie文件
touch .erlang.cookie
# 写入cookie
echo FXZMCVGLBIXZCDEMMVZQ .erlang.cookie
# 修改cookie文件的权限
chmod 600 .erlang.cookie准备三个目录,mq1、mq2、mq3
cd /tmp
# 创建目录
mkdir mq1 mq2 mq3然后拷贝rabbitmq.conf、cookie文件到mq1、mq2、mq3
# 进入/tmp
cd /tmp
# 拷贝
cp rabbitmq.conf mq1
cp rabbitmq.conf mq2
cp rabbitmq.conf mq3
cp .erlang.cookie mq1
cp .erlang.cookie mq2
cp .erlang.cookie mq34.启动集群
创建一个网络
docker network create mq-netdocker volume create
运行命令
docker run -d --net mq-net \
-v ${PWD}/mq1/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf \
-v ${PWD}/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie \
-e RABBITMQ_DEFAULT_USERitcast \
-e RABBITMQ_DEFAULT_PASS123321 \
--name mq1 \
--hostname mq1 \
-p 8071:5672 \
-p 8081:15672 \
rabbitmq:3.8-managementdocker run -d --net mq-net \
-v ${PWD}/mq2/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf \
-v ${PWD}/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie \
-e RABBITMQ_DEFAULT_USERitcast \
-e RABBITMQ_DEFAULT_PASS123321 \
--name mq2 \
--hostname mq2 \
-p 8072:5672 \
-p 8082:15672 \
rabbitmq:3.8-managementdocker run -d --net mq-net \
-v ${PWD}/mq3/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf \
-v ${PWD}/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie \
-e RABBITMQ_DEFAULT_USERitcast \
-e RABBITMQ_DEFAULT_PASS123321 \
--name mq3 \
--hostname mq3 \
-p 8073:5672 \
-p 8083:15672 \
rabbitmq:3.8-management5.测试
在mq1这个节点上添加一个队列
如图在mq2和mq3两个控制台也都能看到
6.数据共享测试
点击这个队列进入管理页面
然后利用控制台发送一条消息到这个队列
结果在mq2、mq3上都能看到这条消息 7.可用性测试
我们让其中一台节点mq1宕机
docker stop mq1然后登录mq2或mq3的控制台发现simple.queue也不可用了
说明数据并没有拷贝到mq2和mq3。
4.3.镜像集群
4.3.1.集群结构和特征
镜像集群本质是主从模式具备下面的特征
交换机、队列、队列中的消息会在各个mq的镜像节点之间同步备份。创建队列的节点被称为该队列的主节点备份到的其它节点叫做该队列的镜像节点。一个队列的主节点可能是另一个队列的镜像节点所有操作都是主节点完成然后同步给镜像节点主宕机后镜像节点会替代成新的主
结构如图
4.3.2.部署
参考课前资料《RabbitMQ部署指南.md》 镜像模式
在刚刚的案例中一旦创建队列的主机宕机队列就会不可用。不具备高可用能力。如果要解决这个问题必须使用官方提供的镜像集群方案。
官方文档地址https://www.rabbitmq.com/ha.html
1.镜像模式的特征
默认情况下队列只保存在创建该队列的节点上。而镜像模式下创建队列的节点被称为该队列的主节点队列还会拷贝到集群中的其它节点也叫做该队列的镜像节点。
但是不同队列可以在集群中的任意节点上创建因此不同队列的主节点可以不同。甚至一个队列的主节点可能是另一个队列的镜像节点。
用户发送给队列的一切请求例如发送消息、消息回执默认都会在主节点完成如果是从节点接收到请求也会路由到主节点去完成。镜像节点仅仅起到备份数据作用。
当主节点接收到消费者的ACK时所有镜像都会删除节点中的数据。
总结如下
镜像队列结构是一主多从从就是镜像所有操作都是主节点完成然后同步给镜像节点主宕机后镜像节点会替代成新的主如果在主从同步完成前主就已经宕机可能出现数据丢失不具备负载均衡功能因为所有操作都会有主节点完成但是不同队列其主节点可以不同可以利用这个提高吞吐量
2.镜像模式的配置
镜像模式的配置有3种模式
ha-modeha-params效果准确模式exactly队列的副本量count集群中队列副本主服务器和镜像服务器之和的数量。count如果为1意味着单个副本即队列主节点。count值为2表示2个副本1个队列主和1个队列镜像。换句话说count 镜像数量 1。如果群集中的节点数少于count则该队列将镜像到所有节点。如果有集群总数大于count1并且包含镜像的节点出现故障则将在另一个节点上创建一个新的镜像。all(none)队列在群集中的所有节点之间进行镜像。队列将镜像到任何新加入的节点。镜像到所有节点将对所有群集节点施加额外的压力包括网络I / O磁盘I / O和磁盘空间使用情况。推荐使用exactly设置副本数为N / 2 1。nodesnode names指定队列创建到哪些节点如果指定的节点全部不存在则会出现异常。如果指定的节点在集群中存在但是暂时不可用会创建节点到当前客户端连接到的节点。
这里我们以rabbitmqctl命令作为案例来讲解配置语法。
语法示例
3.exactly模式
rabbitmqctl set_policy ha-two ^two\. {ha-mode:exactly,ha-params:2,ha-sync-mode:automatic}rabbitmqctl set_policy固定写法ha-two策略名称自定义^two\.匹配队列的正则表达式符合命名规则的队列才生效这里是任何以two.开头的队列名称{ha-mode:exactly,ha-params:2,ha-sync-mode:automatic}: 策略内容 ha-mode:exactly策略模式此处是exactly模式指定副本数量ha-params:2策略参数这里是2就是副本数量为21主1镜像ha-sync-mode:automatic同步策略默认是manual即新加入的镜像节点不会同步旧的消息。如果设置为automatic则新加入的镜像节点会把主节点中所有消息都同步会带来额外的网络开销
4.模式
rabbitmqctl set_policy ha-all ^all\. {ha-mode:all}ha-all策略名称自定义^all\.匹配所有以all.开头的队列名{ha-mode:all}策略内容 ha-mode:all策略模式此处是all模式即所有节点都会称为镜像节点
5.nodes模式
rabbitmqctl set_policy ha-nodes ^nodes\. {ha-mode:nodes,ha-params:[rabbitnodeA, rabbitnodeB]}rabbitmqctl set_policy固定写法ha-nodes策略名称自定义^nodes\.匹配队列的正则表达式符合命名规则的队列才生效这里是任何以nodes.开头的队列名称{ha-mode:nodes,ha-params:[rabbitnodeA, rabbitnodeB]}: 策略内容 ha-mode:nodes策略模式此处是nodes模式ha-params:[rabbitmq1, rabbitmq2]策略参数这里指定副本所在节点名称
6.测试
我们使用exactly模式的镜像因为集群节点数量为3因此镜像数量就设置为2.
运行下面的命令
docker exec -it mq1 rabbitmqctl set_policy ha-two ^two\. {ha-mode:exactly,ha-params:2,ha-sync-mode:automatic}下面我们创建一个新的队列 在任意一个mq控制台查看队列
7.测试数据共享
给two.queue发送一条消息
然后在mq1、mq2、mq3的任意控制台查看消息 8.测试高可用
现在我们让two.queue的主节点mq1宕机
docker stop mq1查看集群状态
查看队列状态
发现依然是健康的并且其主节点切换到了rabbitmq2上
5.仲裁队列
从RabbitMQ 3.8版本开始引入了新的仲裁队列他具备与镜像队里类似的功能但使用更加方便。
4.4.仲裁队列
4.4.1.集群特征
仲裁队列仲裁队列是3.8版本以后才有的新功能用来替代镜像队列具备下列特征
与镜像队列一样都是主从模式支持主从数据同步使用非常简单没有复杂的配置主从同步基于Raft协议强一致
4.4.2.部署
参考课前资料《RabbitMQ部署指南.md》
仲裁队列 从RabbitMQ 3.8版本开始引入了新的仲裁队列他具备与镜像队里类似的功能但使用更加方便。
1.添加仲裁队列
在任意控制台添加一个队列一定要选择队列类型为Quorum类型。 在任意控制台查看队列
可以看到仲裁队列的 2字样。代表这个队列有2个镜像节点。
因为仲裁队列默认的镜像数为5。如果你的集群有7个节点那么镜像数肯定是5而我们集群只有3个节点因此镜像数量就是3.
2.测试
可以参考对镜像集群的测试效果是一样的。
3.集群扩容
4.加入集群
1启动一个新的MQ容器
docker run -d --net mq-net \
-v ${PWD}/.erlang.cookie:/var/lib/rabbitmq/.erlang.cookie \
-e RABBITMQ_DEFAULT_USERitcast \
-e RABBITMQ_DEFAULT_PASS123321 \
--name mq4 \
--hostname mq5 \
-p 8074:15672 \
-p 8084:15672 \
rabbitmq:3.8-management2进入容器控制台
docker exec -it mq4 bash3停止mq进程
rabbitmqctl stop_app4重置RabbitMQ中的数据
rabbitmqctl reset5加入mq1
rabbitmqctl join_cluster rabbitmq16再次启动mq进程
rabbitmqctl start_app5.增加仲裁队列副本
我们先查看下quorum.queue这个队列目前的副本情况进入mq1容器
docker exec -it mq1 bash执行命令
rabbitmq-queues quorum_status quorum.queue结果 现在我们让mq4也加入进来
rabbitmq-queues add_member quorum.queue rabbitmq4结果
再次查看
rabbitmq-queues quorum_status quorum.queue查看控制台发现quorum.queue的镜像数量也从原来的 2 变成了 3
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: /
文章转载自： http://www.morning.lgtzd.cn.gov.cn.lgtzd.cn http://www.morning.xqkjp.cn.gov.cn.xqkjp.cn http://www.morning.mehrim.com.gov.cn.mehrim.com http://www.morning.drwpn.cn.gov.cn.drwpn.cn http://www.morning.qwbht.cn.gov.cn.qwbht.cn http://www.morning.nyqnk.cn.gov.cn.nyqnk.cn http://www.morning.pmxw.cn.gov.cn.pmxw.cn http://www.morning.xgkxy.cn.gov.cn.xgkxy.cn http://www.morning.hjwxm.cn.gov.cn.hjwxm.cn http://www.morning.jfch.cn.gov.cn.jfch.cn http://www.morning.cxlys.cn.gov.cn.cxlys.cn http://www.morning.stsnf.cn.gov.cn.stsnf.cn http://www.morning.sqmbb.cn.gov.cn.sqmbb.cn http://www.morning.hhboyus.cn.gov.cn.hhboyus.cn http://www.morning.ktbjk.cn.gov.cn.ktbjk.cn http://www.morning.stxg.cn.gov.cn.stxg.cn http://www.morning.jcxyq.cn.gov.cn.jcxyq.cn http://www.morning.bnqcm.cn.gov.cn.bnqcm.cn http://www.morning.pdmsj.cn.gov.cn.pdmsj.cn http://www.morning.kjrlp.cn.gov.cn.kjrlp.cn http://www.morning.hcszr.cn.gov.cn.hcszr.cn http://www.morning.dblgm.cn.gov.cn.dblgm.cn http://www.morning.tqrxm.cn.gov.cn.tqrxm.cn http://www.morning.yrnrr.cn.gov.cn.yrnrr.cn http://www.morning.pltbd.cn.gov.cn.pltbd.cn http://www.morning.hphrz.cn.gov.cn.hphrz.cn http://www.morning.jfmyt.cn.gov.cn.jfmyt.cn http://www.morning.tbqdm.cn.gov.cn.tbqdm.cn http://www.morning.mgzjz.cn.gov.cn.mgzjz.cn http://www.morning.ylkkh.cn.gov.cn.ylkkh.cn http://www.morning.rsfp.cn.gov.cn.rsfp.cn http://www.morning.cfynn.cn.gov.cn.cfynn.cn http://www.morning.bgygx.cn.gov.cn.bgygx.cn http://www.morning.rbrd.cn.gov.cn.rbrd.cn http://www.morning.qwpyf.cn.gov.cn.qwpyf.cn http://www.morning.nngq.cn.gov.cn.nngq.cn http://www.morning.pnmtk.cn.gov.cn.pnmtk.cn http://www.morning.nhzxr.cn.gov.cn.nhzxr.cn http://www.morning.tnfyj.cn.gov.cn.tnfyj.cn http://www.morning.ljbch.cn.gov.cn.ljbch.cn http://www.morning.mzjbz.cn.gov.cn.mzjbz.cn http://www.morning.yzxhk.cn.gov.cn.yzxhk.cn http://www.morning.rpwm.cn.gov.cn.rpwm.cn http://www.morning.blbys.cn.gov.cn.blbys.cn http://www.morning.gsjfn.cn.gov.cn.gsjfn.cn http://www.morning.wnywk.cn.gov.cn.wnywk.cn http://www.morning.lptjt.cn.gov.cn.lptjt.cn http://www.morning.xcjbk.cn.gov.cn.xcjbk.cn http://www.morning.tlrxp.cn.gov.cn.tlrxp.cn http://www.morning.rgmls.cn.gov.cn.rgmls.cn http://www.morning.qhkdt.cn.gov.cn.qhkdt.cn http://www.morning.rcmwl.cn.gov.cn.rcmwl.cn http://www.morning.yodajy.cn.gov.cn.yodajy.cn http://www.morning.plznfnh.cn.gov.cn.plznfnh.cn http://www.morning.xrksf.cn.gov.cn.xrksf.cn http://www.morning.ubpsa.cn.gov.cn.ubpsa.cn http://www.morning.kxqfz.cn.gov.cn.kxqfz.cn http://www.morning.xplng.cn.gov.cn.xplng.cn http://www.morning.nlmm.cn.gov.cn.nlmm.cn http://www.morning.gkdqt.cn.gov.cn.gkdqt.cn http://www.morning.nkkpp.cn.gov.cn.nkkpp.cn http://www.morning.dfwkn.cn.gov.cn.dfwkn.cn http://www.morning.kcfnp.cn.gov.cn.kcfnp.cn http://www.morning.haibuli.com.gov.cn.haibuli.com http://www.morning.pshtf.cn.gov.cn.pshtf.cn http://www.morning.rfdqr.cn.gov.cn.rfdqr.cn http://www.morning.fnczn.cn.gov.cn.fnczn.cn http://www.morning.ylxgw.cn.gov.cn.ylxgw.cn http://www.morning.qpsxz.cn.gov.cn.qpsxz.cn http://www.morning.rhfbl.cn.gov.cn.rhfbl.cn http://www.morning.nrrzw.cn.gov.cn.nrrzw.cn http://www.morning.xfxqj.cn.gov.cn.xfxqj.cn http://www.morning.gbljq.cn.gov.cn.gbljq.cn http://www.morning.qkqgj.cn.gov.cn.qkqgj.cn http://www.morning.fthqc.cn.gov.cn.fthqc.cn http://www.morning.sfnr.cn.gov.cn.sfnr.cn http://www.morning.sgpnz.cn.gov.cn.sgpnz.cn http://www.morning.ykrg.cn.gov.cn.ykrg.cn http://www.morning.glbnc.cn.gov.cn.glbnc.cn http://www.morning.bjndc.com.gov.cn.bjndc.com
