RabbitMQ小节

一、RabbitMQ简介

RabbitQM是一款应用程序对应用程序的通讯方法，基于AMQP协议，用Erlang语言开发，因而安装环境配置之前需要首先安装Erlang语言环境。通过客户端发送消息到队列消息接收这读取消息的机制实现程序之间的解耦。

MQ是消费-产生者模型的一个典型代表，一端往消息队列中不断写入消息，而另外一端则可以读取或者订阅消息队列中的消息

RabbitMQ最初起源于金融系统，用于在分布式系统中的消息转发，在易用性，扩展性，高可用性等方便表现不俗。具体特别包括：

　　1、可靠性

　　　　RabbitMQ使用一些机制来保证可靠性，如持久化，传输确认，发布确认

　　2、灵活的路由

　　　　在消息进入队列之前，通过Exchange来路由消息。对于典型的路由功能，RabbitMQ已经提供了一些内置的Exchange来实现。针对更复杂的路由功能，可以将多个Exchange绑定在一起，也通过插件机制实现自己的Exchange。（Exchange： 指定消息按什么规则，路由到哪个Queue，Message消息先要到达Exchange，在Server中承担着从Produce接收Message的责任。）

　　3、消息集群

　　　　多个RabbitMQ服务器可以组成一个集群，行程一个逻辑Broker（Broker：其实就是接收和分发消息的应用，也就是说RabbitMQ Server就是Message Broker。）

　　4、高可用性

　　　　队列可以在集群中的机器上进行镜像，使得部分节点在出问题的情况下依然可用

　　5、多种协议

　　　　RabbitMQ支持多种消息队列协议，比如STOMP，MQTT等

　　6、多语言客户端

　　　　RabbitMQ几乎支持所有常用语言，比如java，.net，python等

　　7、界面管理

　　　　RabbitMQ提供了一个易用的用户界面，使得用户可以监控消息Broker的许多方面

　　8、跟踪机制

　　　　如果消息异常，RabbitMQ提供了消息跟踪机制，使用者可以找到发生了什么

　　9、插件机制

　　　　RabbitMQ提供了许多插件，来从多方面进行扩展，也可以编写自己的插件

二、RabbitMQ安装与配置

　　Rabbit MQ 是建立在强大的Erlang OTP平台上，因此安装RabbitMQ之前要先安装Erlang。

　　　　erlang：http://www.erlang.org/download.html

　　　　rabbitmq：http://www.rabbitmq.com/download.html

　　　　注意：

　　　　　　a、先安装erlang然后再安装rabbitMQ

　　　　　　b、就是rabbitmq的安装目录中是不能带空格的，但是官方安装包会默认的将我们的程序安装到Program Files下如果带空格，但是客户端无法启动等等一系列问题

　　　　　　c、默认安装的Rabbit MQ 监听端口是：5672

　　1、安装完以后erlang需要手动设置ERLANG_HOME 的系统变量。

　　　　输入：set ERLANG_HOME=C:Program Fileserl8.0

　　　　 

　　2、激活Rabbit MQ's Management Plugin

　　　　使用Rabbit MQ 管理插件，可以更好的可视化方式查看Rabbit MQ 服务器实例的状态，你可以在命令行中使用下面的命令激活。

　　　　输入：rabbitmq-plugins.bat  enable  rabbitmq_management

　　　　 

　　　　同时，我们也使用rabbitmqctl控制台命令（位于 rabbitmq_server-3.6.3sbin>）来创建用户，密码，绑定权限等。

　　3、创建管理用户

　　　　输入：rabbitmqctl.bat add_user zhangweizhong weizhong1988

 　　　　

　　4、设置管理员

　　　　输入：rabbitmqctl.bat set_user_tags zhangweizhong administrator

　　　　 

　　5、设置权限

　　　　输入：rabbitmqctl.bat set_permissions -p / zhangweizhong ".*" ".*" ".*"

 　　　　

　　6、 其他命令

　　　　　　a. 查询用户： rabbitmqctl.bat list_users

　　　　　　b. 查询vhosts： rabbitmqctl.bat list_vhosts

　　　　　　c. 启动RabbitMQ服务: net stop RabbitMQ && net start RabbitMQ

　　7、后台管理

　　后端可视化界面访问地址为http://localhost:15672，并且会默认有测试账号和密码，且具有管理员权限（账号：guest 密码guest）。如果可视化界面可以正常登陆，则说明安装配置成功，否则失败

三、消息模型

　　开始之前我们先来了解下消息模型：

消费者（consumer）订阅某个队列。生产者（producer）创建消息，然后发布到队列（queue）中，队列再将消息发送到监听的消费者。如果消费者没上线则消息在队列中保存，直到上线后才将消息发送给消费者

下面我们我们通过demo来了解RabbitMQ的基本用法。

　　1、消息的发送和接受

　　简单的发送消息和接受消息，消息生产者和消息消费者同时监听一个队列通道，然后消费者发送消息，消费者获得消息

　　并且消息只能有一个消费者进行接收，

　　消息生产者

 static void Main(string[] args)
        {
            //创建连接工厂并初始连接
            var factory = new ConnectionFactory();
            //消息队列地址
            factory.HostName = "localhost";
            //账号
            factory.UserName = "guest";
            //密码
            factory.Password = "guest";
            //创建一个连接
            using (var connection = factory.CreateConnection())
            {
                //创建一个通道
                using (var channel = connection.CreateModel())
                {
                    //创建一个队列 名为hello
                    channel.QueueDeclare("hello", false, false, false, null);

                    string message = "";
                    while (message != "exit")
                    {
                        Console.Write("Please enter the message to be sent:");
                        message = Console.ReadLine();
                        var body = Encoding.UTF8.GetBytes(message);
                        //向名为hello的队列中发送消息发送消息
                        channel.BasicPublish("", "hello", null, body); 
                        Console.WriteLine("set message: {0}", message);
                    }
                }
            }
        }

　　　　消息消费者

　　　　

static void Main(string[] args)
        {
            //创建连接工厂并初始连接
            var factory = new ConnectionFactory();
            //消息队列地址
            factory.HostName = "localhost";
            //账号
            factory.UserName = "guest";
            //密码
            factory.Password = "guest";
            //创建一个连接
            using (var connection = factory.CreateConnection())
            {
                //创建一个通道
                using (var channel = connection.CreateModel())
                { 
                    //创建一个队列 名为hello
                    channel.QueueDeclare("hello", false, false, false, null);//创建一个队列
                    //创建一个消费者
                    var consumer = new QueueingBasicConsumer(channel);
                    //开启消息者与通道、队列关联  监听hello队列
                    channel.BasicConsume("hello", true, consumer);

                    Console.WriteLine(" waiting for message.");
                    while (true)
                    {
                        //接收消息并出列
                        var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();

                        var body = ea.Body;//消息主体
                        var message = Encoding.UTF8.GetString(body);
                        Console.WriteLine("Received {0}", message);
                        if (message == "exit")
                        {
                            Console.WriteLine("exit!");
                            break;
                        }

                    }
                }
            }
        }

　　2、循环调度

 　　在一个工作者还在处理消息，并且没有响应消息之前，不要给他分发新的消息。相反，将这条新的消息发送给下一个不那么忙碌的工作者。

　　简单直白点说，加入当前有2个服务的消费者，当第一条消息过来是，第一个消费者接收下次，当第二个消息来是，第二个消费者接收消息，依次类推

　　客户端消息生产者代码同上，消费者代码增加20行区别

 static void Main(string[] args)
        {
            //创建连接工厂并初始连接
            var factory = new ConnectionFactory();
            //消息队列地址
            factory.HostName = "localhost";
            //账号
            factory.UserName = "guest";
            //密码
            factory.Password = "guest";
            //创建一个连接
            using (var connection = factory.CreateConnection())
            {
                //创建一个通道
                using (var channel = connection.CreateModel())
                { 
                    //创建一个队列 名为hello
                    channel.QueueDeclare("hello", false, false, false, null);//创建一个队列
                    //在一个工作者还在处理消息，并且没有响应消息之前，不要给他分发新的消息。相反，将这条新的消息发送给下一个不那么忙碌的工作者。
                    channel.BasicQos(0, 1, false);

                    //创建一个消费者
                    var consumer = new QueueingBasicConsumer(channel);
                    //开启消息者与通道、队列关联  监听hello队列
                    channel.BasicConsume("hello", true, consumer);

                    Console.WriteLine(" waiting for message.");
                    while (true)
                    {
                        //接收消息并出列
                        var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();

                        var body = ea.Body;//消息主体
                        var message = Encoding.UTF8.GetString(body);
                        Console.WriteLine("Received {0}", message);
                        if (message == "exit")
                        {
                            Console.WriteLine("exit!");
                            break;
                        }

                    }
                }
            }
        }

　　3、消息持久化

　　rabbitMQ支持消息的持久化，防止服务器重启后消息丢失，当服务器重启后，会重新加载之前被持久化的消息，

　　消息持久化也不是100%不丢失的，首先rabbitMQ会先将数据存储在cache中，需要时间将数据写到磁盘上，如果在这个时间段内服务重启或者故障数据还是会丢失的

　　代码是在消息生产者调整，具体是第18行第二个参数设为true，然后增加第20 和22行代码

static void Main(string[] args)
        {
            //创建连接工厂并初始连接
            var factory = new ConnectionFactory();
            //消息队列地址
            factory.HostName = "localhost";
            //账号
            factory.UserName = "guest";
            //密码
            factory.Password = "guest";
            //创建一个连接
            using (var connection = factory.CreateConnection())
            {
                //创建一个通道
                using (var channel = connection.CreateModel())
                {
                    //创建一个队列 名为hello    第二个参数true表明队列持久化
                    channel.QueueDeclare("hello1", true, false, false, null);

                    var properties = channel.CreateBasicProperties();
                    //properties.SetPersistent(true);//这个方法提示过时，不建议使用
                    properties.DeliveryMode = 2;//1表示不持久,2.表示持久化
                    string message = "";
                    while (message != "exit")
                    {
                        Console.Write("Please enter the message to be sent:");
                        message = Console.ReadLine();
                        var body = Encoding.UTF8.GetBytes(message);
                        //向名为hello的队列中发送消息发送消息
                        channel.BasicPublish("", "hello1", null, body); 
                        Console.WriteLine("set message: {0}", message);
                    }
                }
            }
        }

　　4、广播模式

　　生产者和消费者如果在一个队列中，则所有消息消费者都可以接收到当前队列中的消息

　　消息生产者，详解见注释。具体代码18，32行

static void Main(string[] args)
        {
            //创建连接工厂并初始连接
            var factory = new ConnectionFactory();
            //消息队列地址
            factory.HostName = "192.168.1.113";
            //账号
            factory.UserName = "gaoxing";
            //密码
            factory.Password = "123";
            //创建一个连接
            using (var connection = factory.CreateConnection())
            {
                //创建一个通道
                using (var channel = connection.CreateModel())
                {
                    //定义一个交换机，且采用广播类型,并设为持久化
                    channel.ExchangeDeclare("publish", "fanout", true);//定义一个交换机，且采用广播类型,并设为持久化
                    //创建一个队列 名为hello    第二个参数true表明队列持久化
                    channel.QueueDeclare("guangbo", true, false, false, null);

                    var properties = channel.CreateBasicProperties();
                    //properties.SetPersistent(true);//这个方法提示过时，不建议使用
                    properties.DeliveryMode = 2;//1表示不持久,2.表示持久化
                    string message = "";
                    while (message != "exit")
                    {
                        Console.Write("Please enter the message to be sent:");
                        message = Console.ReadLine();
                        var body = Encoding.UTF8.GetBytes(message);
                        //向名为hello的队列中发送消息发送消息
                        channel.BasicPublish("publish", "guangbo", properties, body); //发送消息,这里指定了交换机名称,且routeKey会被忽略
                        Console.WriteLine("set message: {0}", message);
                    }
                }
            }
        }

　　消息消费者

　　代码17,20 行

 static void Main(string[] args)
        {
            //创建连接工厂并初始连接
            var factory = new ConnectionFactory();
            //消息队列地址
            factory.HostName = "192.168.1.113";
            //账号
            factory.UserName = "gaoxing";
            //密码
            factory.Password = "123";
            //创建一个连接
            using (var connection = factory.CreateConnection())
            {
                //创建一个通道
                using (var channel = connection.CreateModel())
                {
                    channel.ExchangeDeclare("publish", "fanout", true);//定义一个交换机，且采用广播类型，并持久化该交换机,并设为持久化
                    //创建一个队列 名为hello
                    channel.QueueDeclare("guangbo", true, false, false, null);//创建一个队列
                    channel.QueueBind("guangbo", "publish", "");//将队列绑定到名publish的交换机上，实现消息订阅
                    //在一个工作者还在处理消息，并且没有响应消息之前，不要给他分发新的消息。相反，将这条新的消息发送给下一个不那么忙碌的工作者。
                    channel.BasicQos(0, 1, false);

                    //创建一个消费者
                    var consumer = new QueueingBasicConsumer(channel);
                    //开启消息者与通道、队列关联  监听hello队列
                    channel.BasicConsume("guangbo", true, consumer);

                    Console.WriteLine(" waiting for message.");
                    while (true)
                    {
                        //接收消息并出列
                        var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();

                        var body = ea.Body;//消息主体
                        var message = Encoding.UTF8.GetString(body);
                        Console.WriteLine("Received {0}", message);
                        if (message == "exit")
                        {
                            Console.WriteLine("exit!");
                            break;
                        }

                    }
                }
            }
        }

　　5、消息订阅

　　订阅后则可接收消息，未订阅则无法收到消息

　　消息生产者

 static void Main(string[] args)
        {
            //创建连接工厂并初始连接
            var factory = new ConnectionFactory();
            //消息队列地址
            factory.HostName = "192.168.1.113";
            //账号
            factory.UserName = "gaoxing";
            //密码
            factory.Password = "123";
            //创建一个连接
            using (var connection = factory.CreateConnection())
            {
                //创建一个通道
                using (var channel = connection.CreateModel())
                {
                    //定义一个交换机，且采用广播类型,并持久化该交换机
                    channel.ExchangeDeclare("publish-topic", "topic", true);
                    //创建一个队列,第2个参数为true表示为持久队列
                    channel.QueueDeclare("hello-mq", true, false, false, null);

                    var properties = channel.CreateBasicProperties();
                    //properties.SetPersistent(true);//这个方法提示过时，不建议使用
                    properties.DeliveryMode = 2;//1表示不持久,2.表示持久化
                    string message = "";
                    while (message != "exit")
                    {
                        Console.Write("Please enter the message to be sent:");
                        message = Console.ReadLine();
                        var body = Encoding.UTF8.GetBytes(message);
                        //向名为hello的队列中发送消息发送消息
                        channel.BasicPublish("publish-topic", "hello.test", properties, body); //发送消息,这里指定了交换机名称,且routeKey会被忽略
                        Console.WriteLine("set message: {0}", message);
                    }
                }
            }
        }

　　消息消费者

 static void Main(string[] args)
        {
            //创建连接工厂并初始连接
            var factory = new ConnectionFactory();
            //消息队列地址
            factory.HostName = "192.168.1.113";
            //账号
            factory.UserName = "gaoxing";
            //密码
            factory.Password = "123";
            //创建一个连接
            using (var connection = factory.CreateConnection())
            {
                //创建一个通道
                using (var channel = connection.CreateModel())
                {
                    channel.ExchangeDeclare("publish-topic", "topic", true);//定义一个交换机，且采用广播类型，并持久化该交换机
                    string queueName = channel.QueueDeclare("hello-mq", true, false, false, null);//创建一个队列,第2个参数为true表示为持久队列
                    channel.QueueBind(queueName, "publish-topic", "*.test");//将队列绑定到路由上，实现消息订阅
                    //在一个工作者还在处理消息，并且没有响应消息之前，不要给他分发新的消息。相反，将这条新的消息发送给下一个不那么忙碌的工作者。
                    channel.BasicQos(0, 1, false);

                    //创建一个消费者
                    var consumer = new QueueingBasicConsumer(channel);
                    //开启消息者与通道、队列关联  hello-mq
                    channel.BasicConsume("hello-mq", true, consumer);

                    Console.WriteLine(" waiting for message.");
                    while (true)
                    {
                        //接收消息并出列
                        var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();

                        var body = ea.Body;//消息主体
                        var message = Encoding.UTF8.GetString(body);
                        Console.WriteLine("Received {0}", message);
                        if (message == "exit")
                        {
                            Console.WriteLine("exit!");
                            break;
                        }

                    }
                }
            }
        }

四、Exchange

　　1、fanout

　　2、direct

　　3、topic

　　4、header

五、RPC

六、总结

基于上面的demo和对几种不同exchange路由机制的学习，我们发现RabbitMQ主要是涉及到以下几个核心概念：

1. Publisher：生产者，消息的发送方。
2. Connection：网络连接。
3. Channel：信道，多路复用连接中的一条独立的双向数据流通道。
4. Exchange：交换器（路由器），负责消息的路由到相应队列。
5. Binding：队列与交换器间的关联绑定。消费者将关注的队列绑定到指定交换器上，以便Exchange能准确分发消息到指定队列。
6. Queue：队列，消息的缓冲存储区。
7. Virtual Host：虚拟主机，虚拟主机提供资源的逻辑分组和分离。包含连接，交换，队列，绑定，用户权限，策略等。
8. Broker：消息队列的服务器实体。
9. Consumer：消费者，消息的接收方。

参考地址http://www.cnblogs.com/sheng-jie/p/7192690.html

https://www.cnblogs.com/zhangweizhong/p/5713874.html