新网站多久收录,网站有些什么内容,禹州做网站,网站表单制作文章目录 仲裁队列1、创建交换机2、创建仲裁队列3、验证主节点宕机不影响消息发送和接收 流式队列#xff08;不推荐#xff0c;Kafka主场#xff09;概念 异地容灾一、Federation插件概述 二、Federation交换机1、总体说明2、准备工作3、启用联邦插件4、添加上游连接端点5、… 文章目录 仲裁队列1、创建交换机2、创建仲裁队列3、验证主节点宕机不影响消息发送和接收 流式队列不推荐Kafka主场概念 异地容灾一、Federation插件概述 二、Federation交换机1、总体说明2、准备工作3、启用联邦插件4、添加上游连接端点5、创建控制策略6、测试测试计划创建组件发布消息执行测试 三、Federation队列1、总体说明2、创建控制策略3、测试测试计划创建组件 执行测试 仲裁队列 https://www.rabbitmq.com/docs/quorum-queues 1、创建交换机 和仲裁队列绑定的交换机没有特殊我们还是创建一个direct交换机即可 交换机名称exchange.quorum.test 2、创建仲裁队列 队列名称queue.quorum.test 经典队列类型为Classic 点击队列名称查看详细信息 绑定交换机 3、验证 // 测试是否成功绑定 SpringBootTest public class RabbitMQTest {public static final String EXCHANGE_QUORUM_TEST exchange.quorum.test;public static final String ROUTING_KEY_QUORUM_TEST routing.key.quorum.test;Resourceprivate RabbitTemplate rabbitTemplate;Testpublic void testSendMessageToQuorum() {rabbitTemplate.convertAndSend(EXCHANGE_QUORUM_TEST, ROUTING_KEY_QUORUM_TEST, message test quorum ~~~ );} }基础配置参考 Component Slf4j public class MyProcessor {public static final String QUEUE_QUORUM_TEST queue.quorum.test;RabbitListener(queues {QUEUE_QUORUM_TEST})public void quorumMessageProcess(String data, Message message, Channel channel) throws IOException {log.info(消费端 data);channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);}}主节点宕机不影响消息发送和接收 流式队列不推荐Kafka主场 概念 从客户端支持角度来说生态尚不健全从使用习惯角度来说和原有队列用法不完全兼容从竞品角度来说像Kafka但远远比不上Kafka从应用场景角度来说经典队列适用于系统内部异步通信场景流式队列适用于系统间跨平台、大流量、实时计算场景Kafka主场使用建议Stream队列在目前企业实际应用非常少真有特定场景需要使用肯定会倾向于使用Kafka而不是RabbitMQ Stream未来展望Classic Queue已经有和Quorum Queue合二为一的趋势Stream也有加入进来整合成一种队列的趋势但Stream内部机制决定这很 难 异地容灾 Disaster Recovery at a Different Location Federation插件Shovel插件 概述是通过配置联邦交换机拉取上游节点消息关于联邦交换机的配置都在下游节点配置upstream上游节点为常规配置 一、Federation插件 官方链接 Federation Pluginhttps://www.rabbitmq.com/docs/federation Federated-exchangeshttps://www.rabbitmq.com/docs/federated-exchanges 概述 Federation插件的设计目标是使RabbitMQ在不同的Broker节点之间进行消息传递而无须建立集群。 它可以在不同的管理域中的Broker或集群间传递消息这些管理域可能设置了不同的用户和vhost也可能运行在不同版本的RabbitMQ和Erlang上。Federation基于AMQP 0-9-1协议在不同的Broker之间进行通信并且设计成能够容忍不稳定的网络连接情况。 二、Federation交换机 1、总体说明 各节点操作启用联邦插件下游操作 添加上游连接端点创建控制策略 2、准备工作 为了执行相关测试我们使用Docker创建两个RabbitMQ实例。 特别提示由于Federation机制的最大特点就是跨集群同步数据所以这两个Docker容器中的RabbitMQ实例不加入集群是两个独立的broker实例。 docker run -d \ --name rabbitmq-shenzhen \ -p 51000:5672 \ -p 52000:15672 \ -v rabbitmq-plugin:/plugins \ -e RABBITMQ_DEFAULT_USERguest \ -e RABBITMQ_DEFAULT_PASS123456 \ rabbitmq:3.13-managementdocker run -d \ --name rabbitmq-shanghai \ -p 61000:5672 \ -p 62000:15672 \ -v rabbitmq-plugin:/plugins \ -e RABBITMQ_DEFAULT_USERguest \ -e RABBITMQ_DEFAULT_PASS123456 \ rabbitmq:3.13-management使用 docker ps3、启用联邦插件 在上游、下游节点中都需要开启。 Docker容器中的RabbitMQ已经开启了rabbitmq_federation还需要开启rabbitmq_federation_management rabbitmq-plugins enable rabbitmq_federation rabbitmq-plugins enable rabbitmq_federation_managementrabbitmq_federation_management插件启用后会在Management UI的Admin选项卡下看到 4、添加上游连接端点 在下游节点填写上游节点的连接信息 # # amqp://guest:123456192.168.217.134:51000添加后 5、创建控制策略 6、测试 测试计划 特别提示 普通交换机和联邦交换机名称要一致交换机名称要能够和策略正则表达式匹配上发送消息时两边使用的路由键也要一致队列名称不要求一致 创建组件 交换机名称和路由建键相同队列名称可不同 所在机房交换机名称路由键队列名称深圳机房上游federated.exchange.demorouting.key.demo.testqueue.normal.shenzhen上海机房下游federated.exchange.demorouting.key.demo.testqueue.normal.shanghai 创建组件后可以查看一下联邦状态连接成功的联邦状态如下 查看状态 发布消息执行测试 在上游节点向交换机发布消息 上游 下游 三、Federation队列 1、总体说明 Federation队列和Federation交换机的最核心区别就是 Federation Police作用在交换机上就是Federation交换机Federation Police作用在队列上就是Federation队列 2、创建控制策略 queue.federation ^fed.queue. 3、测试 测试计划 上游节点和下游节点中队列名称是相同的只是下游队列中的节点附加了联邦策略而已 所在机房交换机路由键队列深圳机房上游exchange.normal.shenzhenrouting.key.normal.shenzhenfed.queue.demo上海机房下游————fed.queue.demo 创建组件 上游节点都是常规操作此处省略。重点需要关注的是下游节点的联邦队列创建时需要指定相关参数 创建组件后可以查看一下联邦状态连接成功的联邦状态如下 执行测试 在上游节点向交换机发布消息 注意 但此时发现下游节点中联邦队列并没有接收到消息这是为什么呢这里就体现出了联邦队列和联邦交换机工作逻辑的区别。 对联邦队列来说如果没有监听联邦队列的消费端程序它是不会到上游去拉取消息的 如果有消费端监听联邦队列那么首先消费联邦队列自身的消息如果联邦队列为空这时候才会到上游队列节点中拉取消息。 所以现在的测试效果需要消费端程序配合才能看到 消费端配置 application.yml spring:rabbitmq:host: 192.168.217.134port: 61000username: guestpassword: 123456virtual-host: /listener:simple:acknowledge-mode: manual # 把消息确认模式改为手动确认消费端 Component Slf4j public class MyMessageListener {public static final String QUEUE_NAME fed.queue.demo;RabbitListener(queues {QUEUE_NAME})public void processMessage(String dataString, Message message, Channel channel) throws IOException {log.info([Federation] dataString);channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);}}