RabbitMQ 中文文档
https://www.yuque.com/kevin_study/mq https://www.cnblogs.com/zuowj/p/4971123.html 分布式系统消息中间件

RabbitMQ是什么?

AMQP,即Advanced Message Queuing Protocol,高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计。消息中间件主要用于组件之间的解耦,消息的发送者无需知道消息使用者的存在,反之亦然。
AMQP的主要特征是面向消息、队列、路由(包括点对点和发布/订阅)、可靠性、安全。
RabbitMQ是一个开源的AMQP实现,服务器端用Erlang语言编写,支持多种客户端,如:Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等,支持AJAX。用于在分布式系统中存储转发消息,在易用性、扩展性、高可用性等方面表现不俗。

消息队列的特性

解耦:消息的生产者与消费者均基于AMQP协议(相同的接口与规范)进行发送与接收消息,互相不存依赖;
冗余:消息只有处理了才会被删除,除非明确允许多个消费者可以收到同一消息的多个副本,否则每个消息只会被单个消费者接收并处理;
扩展性:可增加或减少多个消息的生产者与消费者,两者的改动均不会影响到双方;
灵活性 & 峰值处理能力:因为有良好的扩展性,所以可视服务器的处理情况【可称为:消费者】(比如:高并发负载过大)动态的增减服务器,以提提高处理能力(可称为:负载均衡);
可恢复性:消息的生产者与消费者不论哪一方出现问题,均不会影响消息的正常发出与接收(当然单一的生产者与消费者除外,如果是这样也就没有必要使用分布式消息队列);
送达保证:只有消息被确认成功处理后才会被删除,否则会重新分发给其它的消费者进行处理,直到确认处理成功为止;
排序保证:先进先出是队列的基本特性;
缓冲:同一时间有多个消息进入消息队列,但是同一时间可以指定一个多个消息被消息者接收并处理,其余的消息处理等待状态,这样可以降低服务器的压力,起到缓冲的作用;
理解数据流:传递的消息内容以字节数组为主,但可以将对象序列化后成字节数组,然后在消费者接收到消息后,可反序列化成对象并进行相关的处理,应用场景:CQRS;
异步通信:允许将一个或多个消息放入消息队列,但并不立即处理它,而是在恰当的时候再去由一个或多个消费者分别接收并处理它们;
以上是我的个人理解,也可参看《使用消息队列的 10 个理由

应用场景:针对高并发且无需立即返回处理结果的时候,可以考虑使用消息队列,如果处理需要立即返回结果则不适合;

RabbitMQ环境的安装

服务器端:

A.需要先安装Erlang环境,下载地址:http://www.erlang.org/download.html,可能有时无法正常访问,可以通过VPN代理来访问该网站或在其它网站上下载(比如:CSDN)
B.安装RabbitMQ Server(有针对多个操作系统的下载,我这边以WINDOWS平台为主),下载地址:http://www.rabbitmq.com/download.html
说明:最新版的Erlang及abbitMQ Server安装后,一般WINDOWS环境变量及服务均都已正常安装与并正常启动,可不是最新版或没有安装好,则可执行以下命令:
Setx ERLANG_HOME “C:\Program Files\erl7.1″ -Erlang的-安装目录,也可通过系统属性—>高级—>环境变量来手动设置;
cd C:\Program Files (x86)\RabbitMQ Server\rabbitmq_server-3.5.6\sbin —切换到RabbitMQ Server的sbin目录下,然后执行如下命令:

rabbitmq-service install
rabbitmq-service enable
rabbitmq-service start

安装并设置OK后,可以通过:rabbitmqctl status查看运行情况、rabbitmqctl list_users查看当前用户、以下命令增加一个新用户:

rabbitmqctl add_user username password
rabbitmqctl set_permissions username “.“ “.“ “.*”
rabbitmqctl set_user_tags username administrator

修改密码:rabbitmqctl change_password username newpassowrd
删除指定的用户:rabbitmqctl delete_user username
列出所有queue:rabbitmqctl list_queues
列出指定queue的信息:rabbitmqctl list_queues [the queue name] messages_ready messages_unacknowledged
列出所有exchange:rabbitmqctl list_exchanges
列出所有binding:rabbitmqctl list_bindings
安装基于web的管理插件:rabbitmq-plugins.bat enable rabbitmq_management
当然还有其它的命令,大家可以去查看官网及其它资料,但我认为知道以上的命令足够用了

RabbitMQ的基本用法

使用RabbitMQ客户端就必需在项目中引用其相关的组件,这里可以通过NuGet安装或从官网下载再引用均可,方法很简单,不再重述;

普通模式

普通用法:采用默认的exchange(交换机,或称路由器)+默认的exchange类型:direct+noAck(自动应答,接收就应答)

  1. /// <summary>
  2. /// 消息发送者,一般用在客户端
  3. /// </summary>
  4. class RabbitMQPublish
  5. {
  6. static void Main(string[] args)
  7. {
  8. var factory = new ConnectionFactory();//创建连接工厂并初始连接
  9. factory.HostName = "localhost";
  10. factory.UserName = "zwj";
  11. factory.Password = "www.zuowenjun.cn";
  12. using (var connection = factory.CreateConnection())//创建一个连接
  13. {
  14. using (var channel = connection.CreateModel()) //创建一个通道
  15. {
  16. channel.QueueDeclare("hello", false, false, false, null);//创建一个队列
  17. string message = "";
  18. while (message!="exit")
  19. {
  20. Console.Write("Please enter the message to be sent:");
  21. message = Console.ReadLine();
  22. var body = Encoding.UTF8.GetBytes(message);
  23. channel.BasicPublish("", "hello", null, body); //发送消息
  24. Console.WriteLine("set message: {0}", message);
  25. }
  26. }
  27. }
  28. }
  29. }
  1. /// <summary>
  2. /// 消费者,一般用在服务端
  3. /// </summary>
  4. class RabbitMQConsume
  5. {
  6. static void Main(string[] args)
  7. {
  8. var factory = new ConnectionFactory();//创建连接工厂并初始连接
  9. factory.HostName = "localhost";
  10. factory.UserName = "zwj";
  11. factory.Password = "www.zuowenjun.cn";
  12. using (var connection = factory.CreateConnection())//创建一个连接
  13. {
  14. using (var channel = connection.CreateModel())//创建一个通道
  15. {
  16. channel.QueueDeclare("hello", false, false, false, null);//创建一个队列
  17. var consumer = new QueueingBasicConsumer(channel);//创建一个消费者
  18. channel.BasicConsume("hello", true, consumer);//开启消息者与通道、队列关联
  19. Console.WriteLine(" waiting for message.");
  20. while (true)
  21. {
  22. var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();//接收消息并出列
  23. var body = ea.Body;//消息主体
  24. var message = Encoding.UTF8.GetString(body);
  25. Console.WriteLine("Received {0}", message);
  26. if (message == "exit")
  27. {
  28. Console.WriteLine("exit!");
  29. break;
  30. }
  31. }
  32. }
  33. }
  34. }
  35. }

负载均衡模式

负载均衡处理模式:采用默认的exchange(交换机)+智能分发+默认的exchange类型:direct+手动应答
消息生产者/发布者代码与上面相同;
以下是消费者代码:

  1. /// <summary>
  2. /// 消费者,一般用在服务端
  3. /// </summary>
  4. class RabbitMQConsume
  5. {
  6. static void Main(string[] args)
  7. {
  8. var factory = new ConnectionFactory();//创建连接工厂并初始连接
  9. factory.HostName = "localhost";
  10. factory.UserName = "zwj";
  11. factory.Password = "www.zuowenjun.cn";
  12. using (var connection = factory.CreateConnection())//创建一个连接
  13. {
  14. using (var channel = connection.CreateModel())//创建一个通道
  15. {
  16. channel.QueueDeclare("hello", false, false, false, null);//创建一个队列
  17. channel.BasicQos(0, 1, false);//在一个工作者还在处理消息,并且没有响应消息之前,不要给他分发新的消息。相反,将这条新的消息发送给下一个不那么忙碌的工作者。
  18. var consumer = new QueueingBasicConsumer(channel);//创建一个消费者
  19. channel.BasicConsume("hello", false, consumer);//开启消息者与通道、队列关联
  20. Console.WriteLine(" waiting for message.");
  21. while (true)
  22. {
  23. var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();//接收消息并出列
  24. var body = ea.Body;//消息主体
  25. var message = Encoding.UTF8.GetString(body);
  26. Console.WriteLine("Received {0}", message);
  27. channel.BasicAck(ea.DeliveryTag, false);
  28. if (message == "exit")
  29. {
  30. Console.WriteLine("exit!");
  31. break;
  32. }
  33. Thread.Sleep(1000);
  34. }
  35. }
  36. }
  37. }
  38. }

持久化模式

消息持久化模式:在负载均衡处理模式的基础上加上持久化,这样即使生产者或消费者或服务端断开,消息均不会丢失

  1. /// <summary>
  2. /// 消息发送者,一般用在客户端
  3. /// </summary>
  4. class RabbitMQPublish
  5. {
  6. static void Main(string[] args)
  7. {
  8. var factory = new ConnectionFactory();//创建连接工厂并初始连接
  9. factory.HostName = "localhost";
  10. factory.UserName = "zwj";
  11. factory.Password = "www.zuowenjun.cn";
  12. using (var connection = factory.CreateConnection())//创建一个连接
  13. {
  14. using (var channel = connection.CreateModel()) //创建一个通道
  15. {
  16. channel.QueueDeclare("hello", true, false, false, null);//创建一个队列,第2个参数为true表示为持久队列
  17. var properties = channel.CreateBasicProperties();
  18. //properties.SetPersistent(true);这个方法提示过时,不建议使用
  19. properties.DeliveryMode = 2;//1表示不持久,2.表示持久化
  20. string message = "";
  21. while (message!="exit")
  22. {
  23. Console.Write("Please enter the message to be sent:");
  24. message = Console.ReadLine();
  25. var body = Encoding.UTF8.GetBytes(message);
  26. channel.BasicPublish("", "hello", properties, body); //发送消息
  27. Console.WriteLine("set message: {0}", message);
  28. }
  29. }
  30. }
  31. }
  32. }
  1. /// <summary>
  2. /// 消费者,一般用在服务端
  3. /// </summary>
  4. class RabbitMQConsume
  5. {
  6. static void Main(string[] args)
  7. {
  8. var factory = new ConnectionFactory();//创建连接工厂并初始连接
  9. factory.HostName = "localhost";
  10. factory.UserName = "zwj";
  11. factory.Password = "www.zuowenjun.cn";
  12. using (var connection = factory.CreateConnection())//创建一个连接
  13. {
  14. using (var channel = connection.CreateModel())//创建一个通道
  15. {
  16. channel.QueueDeclare("hello", true, false, false, null);//创建一个队列,第2个参数为true表示为持久队列
  17. channel.BasicQos(0, 1, false);//在一个工作者还在处理消息,并且没有响应消息之前,不要给他分发新的消息。相反,将这条新的消息发送给下一个不那么忙碌的工作者。
  18. var consumer = new QueueingBasicConsumer(channel);//创建一个消费者
  19. channel.BasicConsume("hello", false, consumer);//开启消息者与通道、队列关联
  20. Console.WriteLine(" waiting for message.");
  21. while (true)
  22. {
  23. var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();//接收消息并出列
  24. var body = ea.Body;//消息主体
  25. var message = Encoding.UTF8.GetString(body);
  26. Console.WriteLine("Received {0}", message);
  27. channel.BasicAck(ea.DeliveryTag, false);
  28. if (message == "exit")
  29. {
  30. Console.WriteLine("exit!");
  31. break;
  32. }
  33. Thread.Sleep(1000);
  34. }
  35. }
  36. }
  37. }
  38. }

广播订阅模式

广播订阅模式:定义一个交换机,其类型设为广播类型,发送消息时指定这个交换机,消费者的消息队列绑定到该交换机实现消息的订阅,订阅后则可接收消息,未订阅则无法收到消息

  1. /// <summary>
  2. /// 消息发送者/生产者,一般用在客户端
  3. /// </summary>
  4. class RabbitMQPublish
  5. {
  6. static void Main(string[] args)
  7. {
  8. var factory = new ConnectionFactory();//创建连接工厂并初始连接
  9. factory.HostName = "localhost";
  10. factory.UserName = "zwj";
  11. factory.Password = "www.zuowenjun.cn";
  12. using (var connection = factory.CreateConnection())//创建一个连接
  13. {
  14. using (var channel = connection.CreateModel()) //创建一个通道
  15. {
  16. channel.ExchangeDeclare("publish", "fanout",true);//定义一个交换机,且采用广播类型,并设为持久化
  17. string queueName = channel.QueueDeclare("hello", true, false, false, null);//创建一个队列,第2个参数为true表示为持久队列,这里将结果隐式转换成string
  18. var properties = channel.CreateBasicProperties();
  19. //properties.SetPersistent(true);这个方法提示过时,不建议使用
  20. properties.DeliveryMode = 2;//1表示不持久,2.表示持久化
  21. string message = "";
  22. while (message!="exit")
  23. {
  24. Console.Write("Please enter the message to be sent:");
  25. message = Console.ReadLine();
  26. var body = Encoding.UTF8.GetBytes(message);
  27. channel.BasicPublish("publish", "hello", properties, body); //发送消息,这里指定了交换机名称,且routeKey会被忽略
  28. Console.WriteLine("set message: {0}", message);
  29. }
  30. }
  31. }
  32. }
  33. }
  1. /// <summary>
  2. /// 消费者,一般用在服务端
  3. /// </summary>
  4. class RabbitMQConsume
  5. {
  6. static void Main(string[] args)
  7. {
  8. var factory = new ConnectionFactory();//创建连接工厂并初始连接
  9. factory.HostName = "localhost";
  10. factory.UserName = "zwj";
  11. factory.Password = "www.zuowenjun.cn";
  12. using (var connection = factory.CreateConnection())//创建一个连接
  13. {
  14. using (var channel = connection.CreateModel())//创建一个通道
  15. {
  16. channel.ExchangeDeclare("publish", "fanout", true);//定义一个交换机,且采用广播类型,并持久化该交换机,并设为持久化
  17. string queueName = channel.QueueDeclare("hello", true, false, false, null);//创建一个队列,第2个参数为true表示为持久队列
  18. channel.QueueBind(queueName, "publish", "");//将队列绑定到名publish的交换机上,实现消息订阅
  19. channel.BasicQos(0, 1, false);//在一个工作者还在处理消息,并且没有响应消息之前,不要给他分发新的消息。相反,将这条新的消息发送给下一个不那么忙碌的工作者。
  20. var consumer = new QueueingBasicConsumer(channel);//创建一个消费者
  21. channel.BasicConsume(queueName, false, consumer);//开启消息者与通道、队列关联
  22. Console.WriteLine(" waiting for message.");
  23. while (true)
  24. {
  25. var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();//接收消息并出列
  26. var body = ea.Body;//消息主体
  27. var message = Encoding.UTF8.GetString(body);
  28. Console.WriteLine("Received {0}", message);
  29. channel.BasicAck(ea.DeliveryTag, false);//应答
  30. if (message == "exit")
  31. {
  32. Console.WriteLine("exit!");
  33. break;
  34. }
  35. Thread.Sleep(1000);
  36. }
  37. }
  38. }
  39. }
  40. }

主题订阅模式

主题订阅模式:定义一个交换机,其类型设为主题订阅类型,发送消息时指定这个交换机及RoutingKey,消费者的消息队列绑定到该交换机并匹配到RoutingKey实现消息的订阅,订阅后则可接收消息,未订阅则无法收到消息

  1. /// <summary>
  2. /// 消息发送者/生产者,一般用在客户端
  3. /// </summary>
  4. class RabbitMQPublish
  5. {
  6. static void Main(string[] args)
  7. {
  8. var factory = new ConnectionFactory();//创建连接工厂并初始连接
  9. factory.HostName = "localhost";
  10. factory.UserName = "zwj";
  11. factory.Password = "www.zuowenjun.cn";
  12. using (var connection = factory.CreateConnection())//创建一个连接
  13. {
  14. using (var channel = connection.CreateModel()) //创建一个通道
  15. {
  16. channel.ExchangeDeclare("publish-topic", "topic", true);//定义一个交换机,且采用广播类型,并持久化该交换机
  17. channel.QueueDeclare("hello-mq", true, false, false, null);//创建一个队列,第2个参数为true表示为持久队列
  18. var properties = channel.CreateBasicProperties();
  19. //properties.SetPersistent(true);这个方法提示过时,不建议使用
  20. properties.DeliveryMode = 2;//1表示不持久,2.表示持久化
  21. string message = "";
  22. while (message!="exit")
  23. {
  24. Console.Write("Please enter the message to be sent:");
  25. message = Console.ReadLine();
  26. var body = Encoding.UTF8.GetBytes(message);
  27. channel.BasicPublish("publish-topic", "hello.test", properties, body); //发送消息,这里指定了交换机名称,且routeKey会被忽略
  28. Console.WriteLine("set message: {0}", message);
  29. }
  30. }
  31. }
  32. }
  33. }
  1. /// <summary>
  2. /// 消费者,一般用在服务端
  3. /// </summary>
  4. class RabbitMQConsume
  5. {
  6. static void Main(string[] args)
  7. {
  8. var factory = new ConnectionFactory();//创建连接工厂并初始连接
  9. factory.HostName = "localhost";
  10. factory.UserName = "zwj";
  11. factory.Password = "www.zuowenjun.cn";
  12. using (var connection = factory.CreateConnection())//创建一个连接
  13. {
  14. using (var channel = connection.CreateModel())//创建一个通道
  15. {
  16. channel.ExchangeDeclare("publish-topic", "topic",true);//定义一个交换机,且采用广播类型,并持久化该交换机
  17. string queueName = channel.QueueDeclare("hello-mq", true, false, false, null);//创建一个队列,第2个参数为true表示为持久队列
  18. channel.QueueBind(queueName, "publish-topic", "*.test");//将队列绑定到路由上,实现消息订阅
  19. channel.BasicQos(0, 1, false);//在一个工作者还在处理消息,并且没有响应消息之前,不要给他分发新的消息。相反,将这条新的消息发送给下一个不那么忙碌的工作者。
  20. var consumer = new QueueingBasicConsumer(channel);//创建一个消费者
  21. channel.BasicConsume(queueName, false, consumer);//开启消息者与通道、队列关联
  22. Console.WriteLine(" waiting for message.");
  23. while (true)
  24. {
  25. var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();//接收消息并出列
  26. var body = ea.Body;//消息主体
  27. var message = Encoding.UTF8.GetString(body);
  28. Console.WriteLine("Received {0}", message);
  29. channel.BasicAck(ea.DeliveryTag, false);//应答
  30. if (message == "exit")
  31. {
  32. Console.WriteLine("exit!");
  33. break;
  34. }
  35. Thread.Sleep(1000);
  36. }
  37. }
  38. }
  39. }
  40. }

交换机路由类型如下: Direct Exchange:直接匹配,通过Exchange名称+RoutingKey来发送与接收消息; Fanout Exchange:广播订阅,向所有消费者发布消息,但只有消费者将队列绑定到该路由才能收到消息,忽略RoutingKey; Topic Exchange:主题匹配订阅,这里的主题指的是RoutingKey,RoutingKey可以采用通配符,如:或#,RoutingKey命名采用.来分隔多个词,只有消费者将队列绑定到该路由且指定的RoutingKey符合匹配规则时才能收到消息; *Headers Exchange:消息头订阅,消息发布前,为消息定义一个或多个键值对的消息头,然后消费者接收消息时同样需要定义类似的键值对请求头,里面需要多包含一个匹配模式(有:x-mactch=all,或者x-mactch=any),只有请求头与消息头相匹配,才能接收到消息,忽略RoutingKey; 本文内容参考了以下文章: .NET 环境中使用RabbitMQ .Net下RabbitMQ的使用系列文章