RabbitMQ队列初探（一）

一简介

消息队列提供了一个异步通信协议，消息的发送者不用一直等待消息是否被成功处理，而是发完消息立即返回。消息被暂存于队列之中，对消息感兴趣的消费者会订阅消息并处理它们。

消息队列使用场景：

1 ：需要将同一条消息发送给多个感兴趣的系统，这时用消息队列不用把同一条消息分别发送给它们

2：系统的同步处理方式严重影响性能及吞吐量时，如用户订单处理、消息通知等修改为异步处理方式

二服务安装

1 服务环境安装：erlang环境是RabbitMQ运行的前提环境，必须安装。下载并安装最新版Eralng OTP For Windows

2 服务安装：从RabbitMQ官网（http://www.rabbitmq.com/download.html）下载最新安装包安装默认安装的Rabbit MQ 监听端口是5672。

2.1运行命令提示符（管理员权限）定位到RabbitMQ安装目录下

2.2 运行下面命令来启用管理插件：rabbitmq-plugins enable rabbitmq_management

2.3服务重启

rabbitmq-service install //RabbitMq服务安装

rabbitmq-service start //服务启动

2.4打开浏览器，输入http://localhost:15672/回车，会要求输入用户名和密码，用默认的guest/guest

这是RabbitMQ的监控界面，到此RabbitMQ安装成功。

三术语解释

RabbitMQ 是信息传输的中间者。本质上，他从生产者（producers）接收消息，转发这些消息给消费者（consumers）.换句话说，他能够按根据你指定的规则进行消息转发、缓冲、和持久化RabbitMQ 中有一些概念术语，主要包括以下几个：Broker、Exchange、Queue、Binding、Routingkey、Producter、Consumer、Channel。

1.Broker

简单来说就是消息队列服务器的实体。

2.Exchange

接收消息，转发消息到绑定的队列上，指定消息按什么规则，路由到哪个队列。

3.Queue

消息队列载体，用来存储消息，相同属性的 queue 可以重复定义，每个消息都会被投入到一个或多个队列。

4.Binding

绑定，它的作用就是把 Exchange 和 Queue 按照路由规则绑定起来。

5.RoutingKey

路由关键字，Exchange 根据这个关键字进行消息投递。

6.Producter

消息生产者，产生消息的程序。

7.Consumer

消息消费者，接收消息的程序。

8.Channel

消息通道，在客户端的每个连接里可建立多个 Channel，每个 channel 代表一个会话。

四消息传递过程

RabbitMQ消息传递过程：生产者创建消息，然后发布到RabbitMQ服务器。消息的内容包含两部分内容，消息标签和消息体。消息体就是你想要传递的数据。它可以是任何格式的数据，例如JSON数据。消息标签用来描述消息体，RabbitMQ服务器通过它决定由谁获得消息体的内容。消费者连接到RabbitMQ,并订阅到队列（Quenue）上。我们可以把消息队列想象成一个具体的邮箱，每当生产者发送的消息到达邮箱时，RabbitMQ会根据消息标签，把消息体发送到订阅或者监听的消费者。当然消费者接受到消息只是消息的一部分：消息体。在消息的路由中，消息的标签并没有随消息体一块传递。如果消费者需要知道消息是由谁生产的的话，就要看消息体中是否包含发送方（消息生产者）的信息。其实整个过程很简单，生产者生产消息，消费者消费消息。你的应用程序既可以是生产者也可以作为消费者。应用程序（生产者/消费者）和Rabbit代理服务器之间创建一条TCP连接。一旦TCP连接创建，应用程序就可以创建一条AMQP信道。信道是创立在TCP连接之上,RabbitMQ信息都是通过信道发送出去并且每条信道会创建一个唯一ID（RabbitMQ系统会自动记录）。一般不通过TCP连接发送消息，TCP连接的建立或者销毁很费内存。一个TCP连接上可以创建任意多的信道。

五 RabbitMQ(c#) Hello World

1 消息生产者

using System;

using RabbitMQ.Client;

using System.Text;

class Send

{

public static void Main()

{

//设置MabbitMQ所在主机ip或者主机名

var factory = new ConnectionFactory() { HostName = "localhost" };

using(var connection = factory.CreateConnection())//创建一个连接

using(var channel = connection.CreateModel())//创建一个信道

{

channel.QueueDeclare(queue: "hello",durable: false,exclusive: false,autoDelete: false,arguments:null);//创建队列

string message = "Hello World!";//要发送的消息

var body = Encoding.UTF8.GetBytes(message);

channel.BasicPublish(exchange: "",routingKey: "hello",basicProperties: null,

body: body);//发送消息

Console.WriteLine(" [x] Sent {0}", message);

}

Console.WriteLine(" Press [enter] to exit.");

Console.ReadLine();

}

2 消息消费者

class Receive

{

public static void Main()

{

var factory = new ConnectionFactory() { HostName = "localhost" };//设置MabbitMQ所在主机ip或者主机名

using(var connection = factory.CreateConnection())

using(var channel = connection.CreateModel())//声明信道

{

channel.QueueDeclare(queue: "hello",//声明队列

durable: false,

exclusive: false,

autoDelete: false,

arguments: null);

var consumer = new EventingBasicConsumer(channel);//订阅模式 (有消息到达将被自动接收) 消费者

consumer.Received += (model, ea) =>//注册接收事件

{

var body = ea.Body;

var message = Encoding.UTF8.GetString(body);

Console.WriteLine(" [x] Received {0}", message);

};

channel.BasicConsume(queue: "hello",

noAck: true,//接受到消息自动应答

consumer: consumer);