RabbitMQ集群----主备关系,在运行的时候,如果非主要节点宕机,程序操作 不受影响; 如果主节点宕机了, 程序会中断操作。 而Rabbitmq集群,会马上让没有宕机的节点参选,选出新的主要节点。 程序重试的时候,会进入到新的节点中执行。 历史消息不受影响的。
rabbitmq集群">基于Docker构建RabbitMQ集群
1.启动多个RabbitMQ节点
使用Docker启动3个RabbitMQ节点,目标如下表所示:
2.命令启动
节点1:
docker run -d --hostname my-rabbit1 --name rabbit1 -p 5672:5672 -p 15672:15672 -e
RABBITMQ_ERLANG_COOKIE='rabbitcookie' rabbitmq:management
节点2:
docker run -d --hostname my-rabbit2 --name rabbit2 -p 5673:5672 -p 15673:15672 -e
RABBITMQ_ERLANG_COOKIE='rabbitcookie' --link rabbit1:my-rabbit1 rabbitmq:management
节点3:
docker run -d --hostname my-rabbit3 --name rabbit3 -p 5674:5672 -p 15674:15672 -e
RABBITMQ_ERLANG_COOKIE='rabbitcookie' --link rabbit1:my-rabbit1 --link rabbit2:my-rabbit2
rabbitmq:management
注意:由于Erlang节点间通过认证Erlang cookie的方式来允许互相通信,所以 RABBITMQ_ERLANG_COOKIE必须设置为相同的。 启动完成之后,使用docker ps命令查看运行情况,确保RabbitMQ都已经启动。
3.加入集群
内存节点和磁盘节点的选择:
每个RabbitMQ节点,要么是内存节点,要么是磁盘节点。内存节点将所有的队列、交换器、绑定、用 户等元数据定义都存储在内存中;而磁盘节点将元数据存储在磁盘中。单节点系统只允许磁盘类型的节 点,否则当节点重启以后,所有的配置信息都会丢失。如果采用集群的方式,可以选择至少配置一个节 点为磁盘节点,其余部分配置为内存节点,,这样可以获得更快的响应。所以本集群中配置节点1位磁 盘节点,节点2和节点3位内存节点。
集群中的第一个节点将初始元数据代入集群中,并且无须被告知加入。而第2个和之后加入的节点将加 入它并获取它的元数据。要加入节点,需要进入Docker容器,重启RabbitMQ。
设置节点1:
docker exec -it rabbit1 bash
rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl start_app
exit
设置节点2:
docker exec -it rabbit2 bash
rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl join_cluster --ram rabbit@my-rabbit1
rabbitmqctl start_app
exit
设置节点3:
docker exec -it rabbit3 bash
rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl join_cluster --ram rabbit@my-rabbit1
rabbitmqctl start_app
exit
节点设置完成之后,在浏览器访问43.142.250.100:15672、43.142.250.100:15673和 43.142.250.100:15674中任意一个,都会看到RabbitMQ集群已经创建成功。
4.配置镜像队列
镜像队列工作原理:在非镜像队列的集群中,消息会路由到指定的队列。当配置为镜像队列之后,消息 除了按照路由规则投递到相应的队列外,还会投递到镜像队列的拷贝。也可以想象在镜像队列中隐藏着 一个fanout交换器,将消息发送到镜像的队列的拷贝。
进入任意一个RabbitMQ节点,执行如下命令:
rabbitmqctl set_policy ha-all "^" '{"ha-mode":"all"}'
可以设置镜像队列,"^"表示匹配所有队列,即所有队列在各个节点上都会有备份。在集群中,只需要 在一个节点上设置镜像队列,设置操作会同步到其他节点。
rabbitmqctl cluster_status
案例代码:
ConnectionFactory factory = new ConnectionFactory();
factory.AutomaticRecoveryEnabled = true;//如果connection挂掉是否重新连接
factory.TopologyRecoveryEnabled = true;//连接恢复后,连接的交换机,队列等是否一同恢复factory.UserName = UrlConfig.User;//用户名
factory.Password = UrlConfig.Password;//密码 var amqpTcpEndpointList = new List<AmqpTcpEndpoint>()
{new AmqpTcpEndpoint() { HostName="43.142.250.100", Port=5672},new AmqpTcpEndpoint() { HostName="43.142.250.100", Port=5673},new AmqpTcpEndpoint() { HostName="43.142.250.100", Port=5674},
};
using (var connection = factory.CreateConnection(amqpTcpEndpointList))
{using (IModel channel = connection.CreateModel()){#region 声明路由和队列 //支持持久化队列:durable: truechannel.QueueDeclare(queue: "JoinClusterQueue", durable: true,exclusive: false, autoDelete: false, arguments: null);//支持持久化交换机durable: truechannel.ExchangeDeclare(type: ExchangeType.Fanout, exchange: "JoinClusterExChange",durable: true, autoDelete: false, arguments: null);channel.QueueBind(queue: "JoinClusterQueue", exchange: "JoinClusterExChange",routingKey: string.Empty);#endregion//表达发送的是持久化消息var props = channel.CreateBasicProperties();props.Persistent = true;for (int i = 1; true; i++){string msg = $"持久化消息--持久化队列===消息入队确认=={i}";byte[] bytes = Encoding.UTF8.GetBytes(msg);channel.BasicPublish("JoinClusterExChange", string.Empty, props, bytes);Console.WriteLine($"已发送:{msg}");Thread.Sleep(200);}}
}