RabbitMQ(二):理解消息通信RabbitMQ

2023-07-29,,

原文:RabbitMQ(二):理解消息通信RabbitMQ

一、消费者、生产者和信道

  生产者(producer):生产者创建消息,然后发布(发送)到代理服务器(RabbitMQ),可以说发送消息的程序就是生产者。什么是消息?消息包含两部分:有效载荷和标签。有效载荷就是传输的数据,可以是任何内容,包括json数据和图片等等。而标签(一个叫交换器名称和可选的主题标记)描述了有效载荷,RabbitMQ用它来决定谁将获得这个消息。

  消费者(consumer):消费者就是接收消息并处理消息的程序,他们连接到代理服务器上,并订阅到队列上。当消费者接收消息时,它只是得到消息的有效载荷。整个过程很简单:生产者创建消息,消费者接收消息。你的应用程序可以作为生产者也可以作为消费者,在两者之间切换。但是消息的传输必定会通过某一介质传递,此处的消息就通过信道传递。

  信道(channel):不管是发布消息、订阅队列还是接收消息,这些动作都是通过信道完成。应用程序和Rabbit代理服务器之间会创建一条TCP连接,TCP连接就像电缆,信道相当于电缆中的光束。在一条TCP连接上创建多少条信道是没有限制的,所以不会对操作系统的TCP栈造成额外的负担。

二、队列、交换器和绑定

  Rabbit的消息路由分为三部分:交换器、队列和绑定。生产者把消息发布到交换器上;消息最终到达队列,并被消费者接收;绑定决定了消息如何从路由器路由到特定的队列。

  队列(queue):队列是一个栈先进先出,为生产者发布的消息提供了保存的处所,消息在此等待消费。本质上队列可以存储无限的消息,但是需要视系统内存而定。

  绑定(binding):队列通过路由键绑定到交换器,路由键就是消息通过交换器投递到那个队列的规则。

  交换器(exchange):交换器有四种类型:direct、fanout、topic和headers。一个Exchange可以和多个Queue进行绑定,producer在传递消息的时候,会传递一个路由键,Exchange会根据这个路由键按照特定的路由算法,将消息路由给指定的queue。

  (1)direct:如果路由键匹配的话,消息就被投递到对应的队列。类似于单播。

  (2)fanout:将消息广播到绑定的队列上,不管路由键是什么,绑定的队列都会收到消息。

  (3)topic:类似与组播,和正则表达式类似,发送给路由键符合一定规则的队列。如:路由键为user.stock的消息会转发给绑定匹配模式为 * .stock,user.stock, * . * 和#.user.stock.#的队列(* 表是匹配一个任意词组,#表示匹配0个或多个词组)。

  (4)headers:不通过路由键匹配而是通过消息的header匹配,其他与direct交换器一致,但是性能上会差很多。

三、vhost

  每一个RabbitMQ服务器都能创建虚拟消息服务器,我们称之为虚拟主机(vhost)。每一个vhost本质上是一个mini版的RabbitMQ服务器,拥有自己的队列、交换器和绑定,最重要的是拥有自己的权限机制。逻辑上vhost之间是相互独立的分离的,保证了安全性和可移植性。RabbitMQ包含了开箱即用的默认的vhost:"/",因此使用起来非常简单。当在RabbitMQ集群中创建vhost时,整个集群上都会创建该vhost,避免了vhost的重复创建。

四、第一次尝试消息通信

  交换器或队列的时候,如果交换器或队列已经存在,则直接返回结束,不会重复创建。无法承担消息丢失,则生产者和消费者中都需要尝试去创建交换器和队列,如果可以承担则可以由消费者来声明队列。我们创建生产者和消费者两个控制台程序分别运行以下代码。

  生产者:

        static void Main(string[] args)
{
FirstProducer();
} /// <summary>
/// 第一个生产者
/// </summary>
private static void FirstProducer()
{
//1.连接到服务器
var conn_factory = new ConnectionFactory() {
HostName = "localhost",UserName="guest",Password="guest",Port=//默认端口5672
};
using (IConnection conn = conn_factory.CreateConnection())
{
//2.创建信道
using (IModel channel = conn.CreateModel())
{
//3.声明交换器
channel.ExchangeDeclare(
"HelloExchange", //交换器名称
ExchangeType.Direct,//交换器类型
true, //是否持久话
false, //是否自动删除
null //关于交换器的详细设置,键值对形式
);
//4.声明队列
channel.QueueDeclare(
"HelloQueue",//队列名称
false, //是否持久化
false, //是否只对首次声明的队列可见
false, //是否自动删除
null ////关于队列和队列内消息的详细设置,键值对形式
);
//5.绑定交换器和队列
channel.QueueBind(
"HelloQueue", //队列名
"HelloExchange", //交换器名
"hola" //路由键
);
//6.发布消息
string msg_str = "这是生产者第一次发布的消息";
IBasicProperties msg_pro = channel.CreateBasicProperties();
msg_pro.ContentType = "text/plain";//发布的数据类型
for(int i = ; i < ; i++)
{
channel.BasicPublish(
"HelloExchange", //消息发送目标交换器名称
"hola", //路由键
msg_pro, //消息的发布属性
Encoding.UTF8.GetBytes(msg_str) //消息
);
}
}
}
}

  消费者:

        static void Main(string[] args)
{
FirstCousmer();
}
/// <summary>
/// 第一个消费者
/// </summary>
private static void FirstCousmer()
{
//1.链接到服务器
var conn_factory = new ConnectionFactory()
{
HostName = "localhost",UserName = "guest",Password = "guest",Port =
};
using (var conn = conn_factory.CreateConnection())
{
//2.创建信道
using(IModel channel = conn.CreateModel())
{
//3.声明交换器
channel.ExchangeDeclare(
"HelloExchange", //交换器名称
ExchangeType.Direct,//交换器类型
true, //是否持久话
false, //是否自动删除
null //关于交换器的详细设置,键值对形式
);
//4.声明队列
channel.QueueDeclare(
"HelloQueue",//队列名称
false, //是否持久化
false, //是否只对首次声明的队列可见
false, //是否自动删除
null ////关于队列和队列内消息的详细设置,键值对形式
);
//5.绑定交换器和队列
channel.QueueBind(
"HelloQueue", //队列名
"HelloExchange", //交换器名
"hola" //路由键
);
//6.获取消息
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (ch, ea) => //消费者消息接收处理事件
{
var body = Encoding.UTF8.GetString(ea.Body);
Console.WriteLine(body);
channel.BasicAck(ea.DeliveryTag, false); //确认接收消息,从队列中删除
};
//7.启动消费者
string consumer_tag = channel.BasicConsume(
"HelloQueue", //获取的队列名称
false, //是否自动确认接收消息,从队列中删除
consumer //消费者对象
);
channel.BasicCancel(consumer_tag);//调用消费
//var consumer = new QueueingBasicConsumer(channel);
//channel.BasicConsume("HelloQueue", false, consumer);
//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);
//}
}
}
Console.ReadLine();
}

  运行消费者后可以看到,以下结果:

RabbitMQ(二):理解消息通信RabbitMQ的相关教程结束。

《RabbitMQ(二):理解消息通信RabbitMQ.doc》

下载本文的Word格式文档,以方便收藏与打印。