1. 准备3台 centos服务器  192.168.233.128    192.168.233.130    192.168.233.131

防火墙放开 集群端口, 这里一并把所有rabbitmq用到的端口都放开

firewall-cmd --zone=public --add-port=4369/tcp --permanent

firewall-cmd --zone=public --add-port=5671-5672/tcp --permanent

firewall-cmd --zone=public --add-port=25672/tcp --permanent

firewall-cmd --zone=public --add-port=35672-35682/tcp --permanent

firewall-cmd --zone=public --add-port=15672/tcp --permanent

firewall-cmd --zone=public --add-port=61613-61614/tcp --permanent

firewall-cmd --zone=public --add-port=1883/tcp --permanent

firewall-cmd --zone=public --add-port=8883/tcp --permanent

firewall-cmd --zone=public --add-port=15674-15675/tcp --permanent

firewall-cmd --reload

2. 修改host映射  修改 /etc/hosts 文件 ,方便3台机器访问

添加如下  ,添加后重启 3台机器

192.168.233.128 rabbitmq1

192.168.233.130 rabbitmq2

192.168.233.131 rabbitmq3

3. 修改cookie文件, 统一成一个key , [每个erlang程序都一个cookie]

这个cookie 可以在 rabbitmq的日志中 查找到

如图 cookie 文件就在 这个 home dir 目录中

cd /var/lib/rabbitmq

ls -a      --这个是查找目录下所有文件. cookie文件默认是隐藏的

用这个 .erlang.cookie文件中的 key 替换掉 rabbitmq 2,3 中的key ,使 3 台服务器的 .erlang.cookie中的内容一致

4. 集群搭建  参考官网 https://www.rabbitmq.com/clustering.html

根据官网介绍: 我们把rabbitmq2 加入到 rabbitmq1的集群中

在 rabbitmq2 上 停止 rabbitmq

rabbitmqctl stop_app

rabbitmqctl join_cluster rabbit@rabbitmq1

rabbitmqctl start_app

查看 集群状态

rabbitmqctl cluster_status

发现 nodes节点已经变成2个了, 同理 把rabbitmq3加入 rabbitmq1集群中

5. 集群搭建完成后, 添加rabbitmq用户

rabbitmqctl add_user test test //添加用户，后面两个参数分别是用户名和密码，我这都用test了。

rabbitmqctl set_permissions -p / test ".*" ".*" ".*" //添加权限

rabbitmqctl set_user_tags test administrator //修改用户角色

然后登录 ui页面 http://192.168.233.128:15672/#/

发现三个 info中 都是disc 都是磁盘节点

如果想 把节3点改成 内存节点 , 参考官网 http://www.rabbitmq.com/clustering.html

在 rabbitmq3 上执行

rabbitmqctl stop_app

rabbitmqctl change_cluster_node_type ram

rabbitmqctl start_app

可以看到 info中 disc变成RAM了

其他集群命令如: 从集群删除 某节点, 参考官网 都有

-----------------------------------------------------------------------------高可用集群-------------------------------------------------------------------------------------

参考: http://www.rabbitmq.com/ha.html

queue 有两种模式  默认   default

　　　　　　　　  镜像    mirror

queue mirror  =>  master / slave   高可用的队列方案

Each mirrored queue consists of one master and one or more mirrors.

Messages published to the queue are replicated to all mirrors

1. 怎么配置

Mirroring parameters are configured using policies.

HA mode: 把queue 同步到哪些机器上 , 这里选用all 一般业务也就3台rabbitmq服务器就差不多了

ha-sync-mode : 异步策略,比如 新加入节点A 到集群中, automatic=自动同步队列数据 到节点A ,manual 就手动

简单测试下  新建policy

新建 queue test1

发现 node 显示+2  表示 该队列 当前主节点(Master)是rabbitmq3  ,从节点(Slaver)是 rabbitmq1和2

features 中 显示了 队列使用了 我们刚刚定义的 policy

可以自己做下测试 :  新建 policy,不设置 ha-sync-mode

让 rabbitmq1节点挂掉, 推数据到test1 , 然后让 rabbitmq1 重新启动, 数据会不会自动同步到 rabbitmq1 (不会的, 需要手动同步, 而 ha-sync-mode=automatic就会,)

下面展示 c#中怎么连接 rabbitmq集群

class Producter

    {

        const string exchangeName = "myexchange";

        const string queueName = "test1";

        const string routeKey = "test1";

        public static void Main()

        {

            //创建连接工厂

            ConnectionFactory factory = new ConnectionFactory

            {

                UserName = "test",//用户名

                Password = "test",//密码

                //HostName = "118.24.152.28",//rabbitmq ip

            };

            //创建连接

            var connection = factory.CreateConnection(new List<string>() { "192.168.141.128", "192.168.141.129", "192.168.141.130" });

            //创建通道

            var channel = connection.CreateModel();

            var properties = channel.CreateBasicProperties();

            properties.Persistent = true; //消息持久化

            Console.WriteLine(string.Format("开始时间: {0}", DateTime.Now));

            for (int i = ; i < int.MaxValue; i++)

            {

                try

                {

                    var msg = Encoding.UTF8.GetBytes(string.Format("你好{0},{1} ", i, string.Join(",", Enumerable.Range(, ))));

                    channel.BasicPublish(string.Empty, routeKey, properties, msg);

                    Console.WriteLine(i + " 执行完毕");

                    Thread.Sleep();

                }

                catch (Exception ex)

                {

                    Console.WriteLine(ex.ToString());

                }

            }

            Console.WriteLine(string.Format("结束时间: {0}", DateTime.Now));

            channel.Close();

            connection.Close();

        }

    }