自学内容网 自学内容网

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: true
        channel.QueueDeclare(queue: "JoinClusterQueue", durable: true,
            exclusive: false, autoDelete: false, arguments: null);

        //支持持久化交换机durable: true
        channel.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);
        }
    }
}

原文地址:https://blog.csdn.net/nullcodeworld/article/details/145127709

免责声明:本站文章内容转载自网络资源,如本站内容侵犯了原著者的合法权益,可联系本站删除。更多内容请关注自学内容网(zxcms.com)!