设计的有趣的网站推荐,网站建设及推广文案,网站正在建设中色综合,做网站建设的电话销售消息队列在使用过程中#xff0c;面临着很多实际问题需要思考#xff1a; 1. 消息可靠性 消息从发送#xff0c;到消费者接收#xff0c;会经理多个过程#xff1a; 其中的每一步都可能导致消息丢失#xff0c;常见的丢失原因包括#xff1a;
发送时丢失#xff1a; 生…消息队列在使用过程中面临着很多实际问题需要思考 1. 消息可靠性 消息从发送到消费者接收会经理多个过程 其中的每一步都可能导致消息丢失常见的丢失原因包括
发送时丢失 生产者发送的消息未送达exchange消息送达exchange后未到达queue MQ宕机queue将消息丢失consumer接收到消息后未消费就宕机
针对这些问题RabbitMQ分别给出了解决方案
生产者确认机制mq持久化消费者确认机制失败重试机制
下面我们就通过案例来演示每一个步骤。
创建Demo工程 1.1. 生产者消息确认
RabbitMQ提供了publisher confirm机制来避免消息发送到MQ过程中丢失。这种机制必须给每个消息指定一个唯一ID。消息发送到MQ以后会返回一个结果给发送者表示消息是否处理成功。
返回结果有两种方式
publisher-confirm发送者确认 消息成功投递到交换机返回ack消息为投递到交换机返回nack publisher-return发送者回执 消息投递到交换机了但是没有路由到队列返回ack及路由失败原因 注意确认机制发送消息时需要给每个消息设置一个全局唯一id以区分不同消息避免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开启publisher-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 {// 获取RabbitTemplate对象RabbitTemplate rabbitTemplate applicationContext.getBean(RabbitTemplate.class);// 设置ReturnCallbackrabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) - {// 投递失败记录日志log.info(消息发送失败应答码{}原因{}交换机{}路由键{},消息{},replyCode, replyText, exchange, routingKey, message);});}
}1.1.3. 定义ConfirmCallback
ConfirmCallback可以在发送消息时指定因为每个业务处理confirm成功或失败的逻辑不一定相同。
在publisher服务的cn.itcast.mq.spring.SpringAmqpTest类中定义一个单元测试方法
package cn.itcast.mq.spring;import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;import java.util.UUID;Slf4j
RunWith(SpringRunner.class)
SpringBootTest
public class SpringAmqpTest {Autowiredprivate RabbitTemplate rabbitTemplate;Testpublic void testSendMessage2SimpleQueue() throws InterruptedException {String routingKey simple;// 准备消息String message hello, spring amqp!;// 全局唯一的消息ID需要封装到CorrelationData中CorrelationData correlationData new CorrelationData(UUID.randomUUID().toString());// 添加callbackcorrelationData.getFuture().addCallback(result - {assert result ! null;if (result.isAck()) {// ack消息成功log.debug(消息发送成功消息ID{}, correlationData.getId());} else {// nack消息失败log.error(消息发送失败消息ID{}原因:{}, correlationData.getId(), result.getReason());}},ex - log.error(消息发送异常消息ID{}原因:{}, correlationData.getId(), ex.getMessage()));// 发送消息rabbitTemplate.convertAndSend(amq.topic, routingKey, message, correlationData);// 休眠一会儿等待回调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在异常位置打断点再次发送消息程序卡在断点时可以发现此时消息状态为unacked未确定状态 抛出异常后因为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插件
1上传插件
因为我们是基于Docker安装所以需要先查看RabbitMQ的插件目录对应的数据卷。如果不是基于Docker的同学请参考第一章部分重新创建Docker容器。
我们之前设定的RabbitMQ的数据卷名称为mq-plugins所以我们使用下面命令查看数据卷
docker volume inspect mq-plugins可以得到下面结果 接下来将下载的插件上传到这个目录即可
2安装插件
最后就是安装了需要进入MQ容器内部来执行安装。我的容器名为mq所以执行下面命令
docker exec -it mq bash执行时请将其中的 -it 后面的mq替换为你自己的容器名.
进入容器内部后执行下面命令开启插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange结果如下 3使用插件
DelayExchange插件的原理是对官方原生的Exchange做了功能的升级
将DelayExchange接收到的消息暂存在内存中官方的Exchange是无法存储消息的在DelayExchange中计时超时后才投递消息队列中
在RabbitMQ的管理平台声明一个DelayExchange 消息的延迟时间需要在发送消息的时候指定 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 命令解读
rabbitmqctlRabbitMQ的命令行工具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. 部署
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 mq3准备集群配置
在/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. 部署
在刚刚的案例中一旦创建队列的主机宕机队列就会不可用。不具备高可用能力。如果要解决这个问题必须使用官方提供的镜像集群方案。
官方文档地址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命令作为案例来讲解配置语法。
语法示例
1. 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则新加入的镜像节点会把主节点中所有消息都同步会带来额外的网络开销
2. all模式
rabbitmqctl set_policy ha-all ^all\. {ha-mode:all}ha-all策略名称自定义^all\.匹配所有以all.开头的队列名{ha-mode:all}策略内容 ha-mode:all策略模式此处是all模式即所有节点都会称为镜像节点
3. 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]策略参数这里指定副本所在节点名称
3测试
我们使用exactly模式的镜像因为集群节点数量为3因此镜像数量就设置为2.
运行下面的命令
docker exec -it mq1 rabbitmqctl set_policy ha-two ^two\. {ha-mode:exactly,ha-params:2,ha-sync-mode:automatic}下面我们创建一个新的队列 在任意一个mq控制台查看队列 4测试数据共享
给two.queue发送一条消息 然后在mq1、mq2、mq3的任意控制台查看消息 5测试高可用
现在我们让two.queue的主节点mq1宕机
docker stop mq1查看集群状态 查看队列状态 发现依然是健康的并且其主节点切换到了rabbitmq2上。
4.4. 仲裁队列
4.4.1. 集群特征
仲裁队列仲裁队列是3.8版本以后才有的新功能用来替代镜像队列具备下列特征
与镜像队列一样都是主从模式支持主从数据同步使用非常简单没有复杂的配置主从同步基于Raft协议强一致
4.4.2. 部署
从RabbitMQ 3.8版本开始引入了新的仲裁队列他具备与镜像队里类似的功能但使用更加方便。
1 添加仲裁队列
在任意控制台添加一个队列一定要选择队列类型为Quorum类型。 在任意控制台查看队列 可以看到仲裁队列的 2字样。代表这个队列有2个镜像节点。
因为仲裁队列默认的镜像数为5。如果你的集群有7个节点那么镜像数肯定是5而我们集群只有3个节点因此镜像数量就是3.
2 测试
可以参考对镜像集群的测试效果是一样的。
3 集群扩容
1. 加入集群
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_app2. 增加仲裁队列副本
我们先查看下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: /
