RabbitMQ 集群之镜像同步

在上个博文中讲到了如果做集群,那么集群是成功了,但是queue是如何存放的呢?消息又是怎么同步呢。

默认的,也就是什么也不配置,直接在某个节点中添加一个queue,那么它仅仅是属于这个节点的。其它节点有的只是它的影子。所以像断线重连、操作恢复是无法做到的,实验证明确实是这样的。声明queue的节点关闭那么是无法再进行发布消息与消费的。这自然失去了集群的意义

image

 

所以default模式一般是不会进行使用的,我们的选择是 镜像节点

queue mirror 也就是说,也就是 会有一个 master 对应零个或多个 slave

也可以说成主从复制,每个节点都有着相同的数据,某个节点挂掉了,另一个可以立即顶上

 

官方的文档也可以看到,镜像的配置是通过 policy 策略的方式

image

我们有三种同步的方式,一般来说  all 是我们的最佳选择

all 所有的节点都将被同步

exactly 指定个数的节点被同步

nodes 指定的名称的节点被同步

一个简单的示例,下面使用的是ha-all 是我们的策略名称,后面的是^ha\. 则是一个正则,queue名满足这个规则的,将被做镜像。方式是所有的节点都被同步

这里需要说明的是 策略 可以被应用到exchange与queue上面,也可以选择只被应用到哪个上面,下面的命令则是exchange与queue都被应用到了

rabbitmqctl set_policy ha-all "^ha\." "{""ha-mode"":""all""}"

image

翻看文档,似乎没有提供apply to 的参数,所以下面的操作都将使用WEB UI可以看到WEB UI的创建策略给我们提供了丰富的选项,比如Apply to,优先级,等其它设置

image

我们需要关注提上图中 关于 HA的参数

Ha mode 同步方式

Ha params 选择其它同步方式的参数 比如节点名

Ha async mode 同步方式,这个需要详细说明的,默认的情况下。当节点断线后那么这个节点就已经是落后的版本,当我们再去启动节点的时候数据我们需要去手动的同步,这自然是不好的,做到自动化是最完美的,所以设置成automatic是最佳选择

下面设置一个作用与queue,集群中的所有节点将被同步,并且断线重启后自动同步的的策略

image

 

设置完毕后,就可以通过代码来看看效果了,因为使用了集群,所以创建connection的时候需要做些变化了,创建连接的时候把节点的ip放了进去。

AutomaticRecoveryEnabled  断线重连,也就是如果当前的连接断开了,将会尝试重连

TopologyRecoveryEnabled 重连后恢复当前的工作进程,比如channel、queue、发布的消息进度等

               var factory = new ConnectionFactory()
                {
                    UserName = "admin",
                    Password = "admin",
                    AutomaticRecoveryEnabled = true,
                    TopologyRecoveryEnabled = true
                };


                Connection = factory.CreateConnection(new string[3] { "192.168.1.115", "192.168.1.113", "192.168.1.123" });

这里声明了一个test1queue,然后进行发布消息,当节点崩溃,代码自然也会异常。所以try catch起来,但是没关系。因为我们已经设置了断线重连,所以一会儿,就会恢复如常

//创建返回一个新的频道
            using (var channel = RabbitMqHelper.GetConnection().CreateModel())
            {
                channel.QueueDeclare("test1", true, false, false, null);

                var properties = channel.CreateBasicProperties();
                properties.Persistent = true;
                for (var i = 0; i < 10000; i++)
                {
                    try
                    {

                        channel.BasicPublish(string.Empty, "test1", properties, Encoding.UTF8.GetBytes($"这是{i}个消息"));
                        Thread.Sleep(1000);
                        Console.WriteLine($"发布消息 {i}");
                    }
                    catch (Exception ex)
                    {
                        Console.WriteLine(ex.Message);
                        i--;
                    }
                }

            }

程序运行后,在WEB UI中已经可以看到了,主节点是rabbimq1,剩下的是从节点。image

可以看到现在连接到的节点是rabbitmq2,现在我们可以手动的把它挂掉

image

D4UIV_$Y2L]VUDC8Y{[3W~8

可以看到应用程序一直在报错,就这么报了一会,然后又继续发布消息了。这样就保证了我们消息队列的高可用、高可靠

image

本站部分文章源于互联网,本着传播知识、有益学习和研究的目的进行的转载,为网友免费提供。如有著作权人或出版方提出异议,本站将立即删除。如果您对文章转载有任何疑问请告之我们,以便我们及时纠正。

PS:推荐一个微信公众号: askHarries 或者qq群:474807195,里面会分享一些资深架构师录制的视频录像:有Spring,MyBatis,Netty源码分析,高并发、高性能、分布式、微服务架构的原理,JVM性能优化这些成为架构师必备的知识体系。还能领取免费的学习资源,目前受益良多

转载请注明原文出处:Harries Blog™ » RabbitMQ 集群之镜像同步

赞 (0)
分享到:更多 ()

评论 0

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址