标签：wait 发布者 版本 使用 路由器 class sleep 智能 时间
AMQP ，即Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。
AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。
RabbitMQ是一个开源的AMQP实现，服务器端用Erlang语言编写，支持多种客户端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。
安装
Rabbit MQ 是建立在强大的Erlang OTP平台上，因此安装RabbitMQ之前要先安装Erlang。
erlang：http://www.erlang.org/download.html
rabbitmq：http://www.rabbitmq.com/download.html
注意：
1.现在先别装最新的 3.6.3 ，本人在安装完最新的版本，queue 队列有问题，降到了 3.6.2 就解决了。
2.默认安装的Rabbit MQ 监听端口是：5672
|
Protocol |
Bound to |
Port |
|
amqp |
0.0.0.0 |
5672 |
|
amqp |
:: |
5672 |
|
clustering |
:: |
25672 |
|
Context |
Bound to |
Port |
SSL |
Path |
|
RabbitMQ Management |
0.0.0.0 |
15672 |
○ |
/ |
安装完成后，在RabbitMQ的安装目录的sbin会有：rabbitmq-server.bat
在cmd下（以管理员方式运行）：进入sbin目录，运行rabbitmq-server start
监控配置
在cmd下（以管理员方式运行）：进入sbin目录，运行rabbitmq-plugins.bat enable rabbitmq_management
安装完成之后以管理员身份启动
浏览器访问http://localhost:15672 默认账号：guest 密码：guest
消息队列的特性
解耦：消息的生产者与消费者均基于AMQP协议（相同的接口与规范）进行发送与接收消息，互相不存依赖；
冗余：消息只有处理了才会被删除，除非明确允许多个消费者可以收到同一消息的多个副本，否则每个消息只会被单个消费者接收并处理；
扩展性：可增加或减少多个消息的生产者与消费者，两者的改动均不会影响到双方；
灵活性 & 峰值处理能力：因为有良好的扩展性，所以可视服务器的处理情况【可称为：消费者】（比如：高并发负载过大）动态的增减服务器，以提提高处理能力(可称为：负载均衡)；
可恢复性：消息的生产者与消费者不论哪一方出现问题，均不会影响消息的正常发出与接收（当然单一的生产者与消费者除外，如果是这样也就没有必要使用分布式消息队列）；
送达保证：只有消息被确认成功处理后才会被删除，否则会重新分发给其它的消费者进行处理，直到确认处理成功为止；
排序保证：先进先出是队列的基本特性；
缓冲：同一时间有多个消息进入消息队列，但是同一时间可以指定一个多个消息被消息者接收并处理，其余的消息处理等待状态，这样可以降低服务器的压力，起到缓冲的作用；
理解数据流：传递的消息内容以字节数组为主，但可以将对象序列化后成字节数组，然后在消费者接收到消息后，可反序列化成对象并进行相关的处理，应用场景：CQRS；
异步通信：允许将一个或多个消息放入消息队列，但并不立即处理它，而是在恰当的时候再去由一个或多个消费者分别接收并处理它们；
应用场景：针对高并发且无需立即返回处理结果的时候，可以考虑使用消息队列，如果处理需要立即返回结果则不适合；
RabbitMQ的基本用法
使用RabbitMQ客户端就必需在项目中引用其相关的组件，这里可以通过NuGet安装或从官网下载再引用均可，方法很简单，不再重述；
1.普通用法：采用默认的exchange（交换机，或称路由器）+默认的exchange类型：direct+noAck(自动应答，接收就应答)
/// <summary>
/// 消息发送者，一般用在客户端
/// </summary>
class RabbitMQPublish
{
static void Main(string[] args)
{
var factory = new ConnectionFactory();//创建连接工厂并初始连接
factory.HostName = "localhost";
factory.UserName = "zwj";
factory.Password = "www.zuowenjun.cn";
using (var connection = factory.CreateConnection())//创建一个连接
{
using (var channel = connection.CreateModel()) //创建一个通道
{
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);
channel.BasicPublish("", "hello", null, body); //发送消息
Console.WriteLine("set message: {0}", message);
}
}
}
}
}
/// <summary>
/// 消费者，一般用在服务端
/// </summary>
class RabbitMQConsume
{
static void Main(string[] args)
{
var factory = new ConnectionFactory();//创建连接工厂并初始连接
factory.HostName = "localhost";
factory.UserName = "zwj";
factory.Password = "www.zuowenjun.cn";
using (var connection = factory.CreateConnection())//创建一个连接
{
using (var channel = connection.CreateModel())//创建一个通道
{
channel.QueueDeclare("hello", false, false, false, null);//创建一个队列
var consumer = new QueueingBasicConsumer(channel);//创建一个消费者
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.负载均衡处理模式：采用默认的exchange（交换机）+智能分发+默认的exchange类型：direct+手动应答
消息生产者/发布者代码与上面相同；
以下是消费者代码：
/// <summary>
/// 消费者，一般用在服务端
/// </summary>
class RabbitMQConsume
{
static void Main(string[] args)
{
var factory = new ConnectionFactory();//创建连接工厂并初始连接
factory.HostName = "localhost";
factory.UserName = "zwj";
factory.Password = "www.zuowenjun.cn";
using (var connection = factory.CreateConnection())//创建一个连接
{
using (var channel = connection.CreateModel())//创建一个通道
{
channel.QueueDeclare("hello", false, false, false, null);//创建一个队列
channel.BasicQos(0, 1, false);//在一个工作者还在处理消息，并且没有响应消息之前，不要给他分发新的消息。相反，将这条新的消息发送给下一个不那么忙碌的工作者。
var consumer = new QueueingBasicConsumer(channel);//创建一个消费者
channel.BasicConsume("hello", false, 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);
channel.BasicAck(ea.DeliveryTag, false);
if (message == "exit")
{
Console.WriteLine("exit!");
break;
}
Thread.Sleep(1000);
}
}
}
}
}
3.消息持久化模式：在2的基础上加上持久化，这样即使生产者或消费者或服务端断开，消息均不会丢失
/// <summary>
/// 消息发送者，一般用在客户端
/// </summary>
class RabbitMQPublish
{
static void Main(string[] args)
{
var factory = new ConnectionFactory();//创建连接工厂并初始连接
factory.HostName = "localhost";
factory.UserName = "zwj";
factory.Password = "www.zuowenjun.cn";
using (var connection = factory.CreateConnection())//创建一个连接
{
using (var channel = connection.CreateModel()) //创建一个通道
{
channel.QueueDeclare("hello", true, false, false, null);//创建一个队列,第2个参数为true表示为持久队列
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);
channel.BasicPublish("", "hello", properties, body); //发送消息
Console.WriteLine("set message: {0}", message);
}
}
}
}
}
/// <summary>
/// 消费者，一般用在服务端
/// </summary>
class RabbitMQConsume
{
static void Main(string[] args)
{
var factory = new ConnectionFactory();//创建连接工厂并初始连接
factory.HostName = "localhost";
factory.UserName = "zwj";
factory.Password = "www.zuowenjun.cn";
using (var connection = factory.CreateConnection())//创建一个连接
{
using (var channel = connection.CreateModel())//创建一个通道
{
channel.QueueDeclare("hello", true, false, false, null);//创建一个队列,第2个参数为true表示为持久队列
channel.BasicQos(0, 1, false);//在一个工作者还在处理消息，并且没有响应消息之前，不要给他分发新的消息。相反，将这条新的消息发送给下一个不那么忙碌的工作者。
var consumer = new QueueingBasicConsumer(channel);//创建一个消费者
channel.BasicConsume("hello", false, 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);
channel.BasicAck(ea.DeliveryTag, false);
if (message == "exit")
{
Console.WriteLine("exit!");
break;
}
Thread.Sleep(1000);
}
}
}
}
}
4.广播订阅模式：定义一个交换机，其类型设为广播类型，发送消息时指定这个交换机，消费者的消息队列绑定到该交换机实现消息的订阅，订阅后则可接收消息，未订阅则无法收到消息
/// <summary>
/// 消息发送者/生产者，一般用在客户端
/// </summary>
class RabbitMQPublish
{
static void Main(string[] args)
{
var factory = new ConnectionFactory();//创建连接工厂并初始连接
factory.HostName = "localhost";
factory.UserName = "zwj";
factory.Password = "www.zuowenjun.cn";
using (var connection = factory.CreateConnection())//创建一个连接
{
using (var channel = connection.CreateModel()) //创建一个通道
{
channel.ExchangeDeclare("publish", "fanout",true);//定义一个交换机，且采用广播类型,并设为持久化
string queueName = channel.QueueDeclare("hello", true, false, false, null);//创建一个队列,第2个参数为true表示为持久队列,这里将结果隐式转换成string
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);
channel.BasicPublish("publish", "hello", properties, body); //发送消息,这里指定了交换机名称,且routeKey会被忽略
Console.WriteLine("set message: {0}", message);
}
}
}
}
}
/// <summary>
/// 消费者，一般用在服务端
/// </summary>
class RabbitMQConsume
{
static void Main(string[] args)
{
var factory = new ConnectionFactory();//创建连接工厂并初始连接
factory.HostName = "localhost";
factory.UserName = "zwj";
factory.Password = "www.zuowenjun.cn";
using (var connection = factory.CreateConnection())//创建一个连接
{
using (var channel = connection.CreateModel())//创建一个通道
{
channel.ExchangeDeclare("publish", "fanout", true);//定义一个交换机，且采用广播类型，并持久化该交换机,并设为持久化
string queueName = channel.QueueDeclare("hello", true, false, false, null);//创建一个队列,第2个参数为true表示为持久队列
channel.QueueBind(queueName, "publish", "");//将队列绑定到名publish的交换机上，实现消息订阅
channel.BasicQos(0, 1, false);//在一个工作者还在处理消息，并且没有响应消息之前，不要给他分发新的消息。相反，将这条新的消息发送给下一个不那么忙碌的工作者。
var consumer = new QueueingBasicConsumer(channel);//创建一个消费者
channel.BasicConsume(queueName, false, 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);
channel.BasicAck(ea.DeliveryTag, false);//应答
if (message == "exit")
{
Console.WriteLine("exit!");
break;
}
Thread.Sleep(1000);
}
}
}
}
}
5.主题订阅模式：定义一个交换机，其类型设为主题订阅类型，发送消息时指定这个交换机及RoutingKey，消费者的消息队列绑定到该交换机并匹配到RoutingKey实现消息的订阅，订阅后则可接收消息，未订阅则无法收到消息
/// <summary>
/// 消息发送者/生产者，一般用在客户端
/// </summary>
class RabbitMQPublish
{
static void Main(string[] args)
{
var factory = new ConnectionFactory();//创建连接工厂并初始连接
factory.HostName = "localhost";
factory.UserName = "zwj";
factory.Password = "www.zuowenjun.cn";
using (var connection = factory.CreateConnection())//创建一个连接
{
using (var channel = connection.CreateModel()) //创建一个通道
{
channel.ExchangeDeclare("publish-topic", "topic", true);//定义一个交换机，且采用广播类型,并持久化该交换机
channel.QueueDeclare("hello-mq", true, false, false, null);//创建一个队列,第2个参数为true表示为持久队列
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);
channel.BasicPublish("publish-topic", "hello.test", properties, body); //发送消息,这里指定了交换机名称,且routeKey会被忽略
Console.WriteLine("set message: {0}", message);
}
}
}
}
}
/// <summary>
/// 消费者，一般用在服务端
/// </summary>
class RabbitMQConsume
{
static void Main(string[] args)
{
var factory = new ConnectionFactory();//创建连接工厂并初始连接
factory.HostName = "localhost";
factory.UserName = "zwj";
factory.Password = "www.zuowenjun.cn";
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);//创建一个消费者
channel.BasicConsume(queueName, false, 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);
channel.BasicAck(ea.DeliveryTag, false);//应答
if (message == "exit")
{
Console.WriteLine("exit!");
break;
}
Thread.Sleep(1000);
}
}
}
}
}
交换机路由类型如下：
Direct Exchange：直接匹配，通过Exchange名称+RoutingKey来发送与接收消息;
Fanout Exchange：广播订阅，向所有消费者发布消息，但只有消费者将队列绑定到该路由才能收到消息，忽略RoutingKey；
Topic Exchange：主题匹配订阅，这里的主题指的是RoutingKey，RoutingKey可以采用通配符，如：*或#，RoutingKey命名采用.来分隔多个词，只有消费者将队列绑定到该路由且指定的RoutingKey符合匹配规则时才能收到消息；
Headers Exchange：消息头订阅，消息发布前，为消息定义一个或多个键值对的消息头，然后消费者接收消息时同样需要定义类似的键值对请求头，里面需要多包含一个匹配模式（有：x-mactch=all,或者x-mactch=any）,只有请求头与消息头相匹配，才能接收到消息，忽略RoutingKey；
标签：wait 发布者 版本 使用 路由器 class sleep 智能 时间
原文地址：http://www.cnblogs.com/feiyu_Team/p/6217027.html
