RabbitMQ先容以及基本使用

文章目次

一、什么是消息队列？
二、消息队列的作用（长处）
1、解耦
2、流量削峰
3、异步
4、顺序性
三、RabbitMQ基本结构
四、RabbitMQ队列模式
1、简单队列模式
2、工作队列模式
3、发布/订阅模式
4、路由模式
5、主题模式
6、RPC模式
7、发布者确认模式
五、RabbitMQ相干属性描述
总结

---

一、什么是消息队列？

消息队列是一种用于在分布式系统中举行通信的技术。它是一种存储和转发消息的中间件，可以用
于将应用步伐之间的通信解耦，从而实现高效的异步通信。消息队列答应发送者将消息发送到队列
中，而接收者则可以从队列中获取消息并举行处理惩罚。这种方式可以资助系统实现高可用性、高性
能、松耦合和可伸缩性。消息队列通常包罗生产者（发送消息的应用步伐）、斲丧者（接收消息的
应用步伐）和队列（存储消息的缓冲区）。
RabbitMQ：是由erlang语言开发，基于AMQP(高级消息队列协议)协议实现的一种消息队列。市面
上还有很多消息队列，比如Kafka、RocketMQ、Redis等，各有优劣，本文主要先容RabbitMQ。
官方文档：RabbitMQ Tutorials | RabbitMQ
二、消息队列的作用（长处）

1、解耦

应用步伐解耦，通过引入消息队列，不同的应用步伐之间可以通过消息队列举行通信，而无需直接
调用对方的接口或方法。这样可以降低系统中各个应用步伐之间的耦合度，使得它们可以独立演化
和扩展，而不会因为对方的变化而受到影响。
2、流量削峰

消息队列可以作为一个缓冲区，临时存储流入的消息，直到系统有足够的资源来处理惩罚它们。当系统
出现流量高峰时，消息队列可以临时存储过多的消息，以平滑处理惩罚流量的颠簸，避免系统被突发的
高负载压垮。
3、异步

发送者在发送消息后可以立即继续实行其他操作，而不需要等候接收者的响应。这样可以进步系统
的并发性和响应速度。也可以资助进步系统的吞吐量，特殊是在面对大量请求或处理惩罚复杂计算时。
发送者可以并行地向多个接收者发送消息，而不会因为等候接收者的响应而阻塞。
4、顺序性

固然并不是所有消息队列都能保证消息的绝对顺序性，但是在很多情况下，消息队列可以保证消息
的相对顺序性。即按照发送顺序举行处理惩罚，对某些场景要求顺序实行很适合。
三、RabbitMQ基本结构

名称
描述
Connection（毗连 ）
毗连是生产者和斲丧者与RabbitMQ之间的毗连。每个生产者和斲丧者都需要与RabbitMQ建立一个毗连，以便发送和接收消息。毗连通常是长毗连，可以重用以进步性能和服从。
Channel（信道）
Channel是毗连（Connection）内的逻辑通道，用于完成大部分 AMQP 操作，如声明队列、发送和接收消息等。在 RabbitMQ 中引入 Channel（信道）的主要目标是为了进步系统的性能、灵活性和服从。使用 Channel 可以避免频仍地创建和烧毁毗连，因为一个毗连可以包含多个 Channel。这样可以减少毗连的开销，节省系统资源，并进步性能。
Exchange（交换机）
交换机是消息的接收和分发中心，负责接收生产者发送的消息，并根据指定的路由规则发送到一个或多个队列中。
（Exchange相称于Queue的代理，可以设置不同的写入计谋，写入到对应的队列中。对于队列的写入，更加灵活）
交换机的类型有：fanout扇出、topic主题、direct直接
Queue（队列）
队列是消息的缓存区，用于存储交换机发送的消息。生产者发送的消息终极会被存储在队列中，等候斲丧者举行斲丧。队列可以长期化到磁盘，以确保消息不会在RabbitMQ宕机或重启后丢失。
Producer（生产者）
生产者是发送消息到RabbitMQ的应用步伐。生产者负责创建消息并将其发送到RabbitMQ的消息队列中。
Consumer（斲丧者）
斲丧者是从RabbitMQ队列中接收消息并举行处理惩罚的应用步伐。斲丧者可以订阅一个或多个队列，并在消息到达队列时接收并处理惩罚它们。斲丧者负责监听队列中的消息，并将其取出举行处理惩罚。
四、RabbitMQ队列模式

基于Exchange交换机，RabbitMQ截至现在有七种队列模式。
1、简单队列模式

一个消息生产者，一个消息斲丧者，一个队列。也称为点对点模式。

图中P代表生产者，C代表斲丧者，Queue是队列名称。
我们看到是没有Exchange的，但是RabbitMQ也会有一个默认的交换机。这个默认的交换机通常被
称为"amq.default"或者""（空字符串），是RabbitMQ自动创建的，用于在没有指定交换机的情况
下将消息发送到队列。

1. //生产者
2. var factory = new ConnectionFactory { HostName = "localhost"}; //初始化连接信息
3. using var connection = factory.CreateConnection();  //创建连接
4. using var channel = connection.CreateModel(); //创建信道
6. //声明一个队列，并将信道与队列绑定
7. channel.QueueDeclare(queue: "hello",
8.                      durable: false,
9.                      exclusive: false,
10.                      autoDelete: false,
11.                      arguments: null);
12. //发送消息的内容
13. string message = $"Hello World!";
14. var body = Encoding.UTF8.GetBytes(message);
16. //信道绑定交换机
17. channel.BasicPublish(exchange: string.Empty,
18.                      routingKey: string.Empty,
19.                      basicProperties: null,
20.                      body: body);
22. Console.WriteLine($" [x] Sent {message}");
24. Console.WriteLine(" Press [enter] to exit.");
27. //消费者
28. var factory = new ConnectionFactory { HostName = "localhost" };
29. using var connection = factory.CreateConnection();
30. using var channel = connection.CreateModel();
32. channel.QueueDeclare(queue: "hello",
33.                      durable: false,
34.                      exclusive: false,
35.                      autoDelete: false,
36.                      arguments: null);
38. Console.WriteLine(" [*] Waiting for messages.");
40. var consumer = new EventingBasicConsumer(channel);
41. consumer.Received += (model, ea) =>
42. {
43.     var body = ea.Body.ToArray();
44.     var message = Encoding.UTF8.GetString(body);
45.     Console.WriteLine($" [x] Received {message}");
46. };
47. channel.BasicConsume(queue: "hello",
48.                      autoAck: true,
49.                      consumer: consumer);
51. Console.WriteLine(" Press [enter] to exit.");

复制代码

此时就会生产者发送一条消息，斲丧者就会接收一条消息。
2、工作队列模式

_工作队列又叫做使命队列，正常_会按顺序把消息发送给每一个订阅的斲丧者，均匀而言，每个斲丧
者将获得雷同数量的消息。（不是P发送一条消息，C1和C2都会收到，而是第一条C1斲丧，第二
条C2斲丧。每个消息只会被一个斲丧者接收和处理惩罚）。
这样的好处是可以进步吞吐量，因为生产者发送了很多消息，但是斲丧者只有一个，斲丧者处理惩罚很
慢，就会造成消息积蓄。

1. //生产者
2. var factory = new ConnectionFactory { HostName = "localhost"};
3. using var connection = factory.CreateConnection();
4. using var channel = connection.CreateModel();
6. channel.QueueDeclare(queue: "task_queue",
7.                      durable: true,
8.                      exclusive: false,
9.                      autoDelete: false,
10.                      arguments: null);
12. var message = $"work queue";
13. var body = Encoding.UTF8.GetBytes(message);
14. channel.BasicPublish(exchange: string.Empty,
15.                  routingKey: string.Empty,
16.                  basicProperties: null,
17.                  body: body);
18. Console.WriteLine($" [x] Sent {message}");
20. Console.WriteLine(" Press [enter] to exit.");
22. //消费者
23. var factory = new ConnectionFactory { HostName = "localhost" };
24. using var connection = factory.CreateConnection();
25. using var channel = connection.CreateModel();
27. channel.QueueDeclare(queue: "task_queue",
28.                      durable: true,
29.                      exclusive: false,
30.                      autoDelete: false,
31.                      arguments: null);
33. Console.WriteLine(" [*] Waiting for messages.");
35. var consumer = new EventingBasicConsumer(channel);
36. consumer.Received += (model, ea) =>
37. {
38.     byte[] body = ea.Body.ToArray();
39.     var message = Encoding.UTF8.GetString(body);
40.     Console.WriteLine($" [x] Received {message}");
42.     channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
43. };
44. channel.BasicConsume(queue: "task_queue",
45.                      autoAck: false,
46.                      consumer: consumer);
48. Console.WriteLine(" Press [enter] to exit.");

复制代码

工作队列与简单队列一致，会有一个默认的交换机。
3、发布/订阅模式

发布/订阅模式是一种消息传递模式，它答应发送者（发布者）将消息发布到多个接收者（订阅
者）。消息传递模型的核心思想是生产者从不直接向队列发送任何消息。实际上，生产者通常根本
不知道消息是否会被传递到任何队列。
以是消息传递模式，发布者不需要指定队列。
发布/订阅模式交换机类型为Fanout。

1. //发布者
2. var factory = new ConnectionFactory { HostName = "localhost"};
3. using var connection = factory.CreateConnection();
4. using var channel = connection.CreateModel();
6. //声明一个交换机，叫做logs，并且交换机的类型是Fanout
7. channel.ExchangeDeclare(exchange: "logs", type: ExchangeType.Fanout);
9. var message = "publish_subscribe";
10. var body = Encoding.UTF8.GetBytes(message);
11. channel.BasicPublish(exchange: "logs",
12.                      routingKey: string.Empty,
13.                      basicProperties: null,
14.                      body: body);
15. Console.WriteLine($" [x] Sent {message}");
17. Console.WriteLine(" Press [enter] to exit.");
19. //接收者
20. var factory = new ConnectionFactory { HostName = "localhost"};
21. using var connection = factory.CreateConnection();
22. using var channel = connection.CreateModel();
24. channel.ExchangeDeclare(exchange: "logs", type: ExchangeType.Fanout);
26. //创建一个具有生成名称的非持久、独占、自动删除队列
27. var queueName = channel.QueueDeclare().QueueName;
28. channel.QueueBind(queue: queueName,
29.                   exchange: "logs",
30.                   routingKey: string.Empty);
32. Console.WriteLine(" [*] Waiting for logs.");
34. var consumer = new EventingBasicConsumer(channel);
35. consumer.Received += (model, ea) =>
36. {
37.     byte[] body = ea.Body.ToArray();
38.     var message = Encoding.UTF8.GetString(body);
39.     Console.WriteLine($" [x] {message}");
40. };
41. channel.BasicConsume(queue: queueName,
42.                      autoAck: false,
43.                      consumer: consumer);
45. Console.WriteLine(" Press [enter] to exit.");

复制代码

注：如果发布者已经发布消息到交换机，但还没有队列绑定到交换机，消息将会丢失。
4、路由模式

路由模式也是一种消息传递模式，是基于消息的路由键（routing key）来将消息从交换机
（exchange）发送到一个或多个队列中。相比较于发布/订阅模式，路由模式多了一个routing key
的概念。
路由模式交换机类型为Direct。

1. //生产者
2. var factory = new ConnectionFactory { HostName = "localhost"};
3. using var connection = factory.CreateConnection();
4. using var channel = connection.CreateModel();
6. //定义交换机名称以及类型为Direct
7. channel.ExchangeDeclare(exchange: "direct_logs", type: ExchangeType.Direct);
9. //定义路由键
10. string routingKey = "direct_test";
12. //发送消息体
13. string message  = "direct_message";
14. var body = Encoding.UTF8.GetBytes(message);
16. channel.BasicPublish(exchange: "direct_logs",
17.                      routingKey: routingKey,
18.                      basicProperties: null,
19.                      body: body);
20. Console.WriteLine($" [x] Sent '{routingKey}':'{message}'");
22. Console.WriteLine(" Press [enter] to exit.");
25. //消费者
26. var factory = new ConnectionFactory { HostName = "localhost" };
27. using var connection = factory.CreateConnection();
28. using var channel = connection.CreateModel();
30. channel.ExchangeDeclare(exchange: "direct_logs", type: ExchangeType.Direct);
32. //创建一个具有生成名称的非持久、独占、自动删除队列
33. var queueName = channel.QueueDeclare().QueueName;
35. //路由键集合
36. var routeKeyArr = new string[] { "direct_test", "direct_test2" };
38. foreach (var routeKey in routeKeyArr)
39. {
40.     channel.QueueBind(queue: queueName,
41.                       exchange: "direct_logs",
42.                       routingKey: routeKey);
43. }
45. Console.WriteLine(" [*] Waiting for messages.");
47. var consumer = new EventingBasicConsumer(channel);
48. consumer.Received += (model, ea) =>
49. {
50.     var body = ea.Body.ToArray();
51.     var message = Encoding.UTF8.GetString(body);
52.     var routingKey = ea.RoutingKey;
53.     Console.WriteLine($" [x] Received '{routingKey}':'{message}'");
54. };
55. channel.BasicConsume(queue: queueName,
56.                      autoAck: true,
57.                      consumer: consumer);
59. Console.WriteLine(" Press [enter] to exit.");

复制代码

路由模式，斲丧者可以监听多个路由键。
5、主题模式

基于路由模式，仍然有范围性——它不能基于多个标准举行路由。也就是一个斲丧者只能接收完全
与routing key相匹配的交换机。主题模式主要解决路由模式的不敷，可以模糊匹配routing key。
路由模式交换机类型为Topic。

在生产者方面，基于 . 作为分隔符，用于routing key。比如“stock.usd.nyse”、“nyse.vmw”、
“quick.orange.rabbit”。可以是任何单词，但最多只有255 个字节。
在斲丧者方面，绑定routing key有两种重要的情况：
（1）*(星号)：匹配一个单词。
详细语法：

1. var routeing_key = "info.debug.error";
3. //匹配 info
4. "info.*.*"
5. //匹配debug
6. "*.debug.*"
7. //匹配error
8. "*.*.error"

复制代码

（2）#(散列)：匹配零个或多个单词。
详细语法：

1. var routeing_key = "info.debug.error";
3. //匹配 info
4. "info.#"
5. //匹配debug
6. "#.debug.#"
7. //匹配error
8. "*.*.error"

复制代码

6、RPC模式

RPC模式又叫"请求/回复模式"。
RPC（Remote Procedure Call，远程过程调用）是一种用于在分布式系统中举行通信的技术。它
答应一个进程（或线程）调用另一个进程（或线程）的过程（函数或方法），就像调用本地函数一
样，而不需要开发者显式处理惩罚底层通信细节。
（就是生产者发送一条消息，斲丧者端实行某个方法，获取值的同时，并返回到生产者。）

1. //生产者
2. var factory = new ConnectionFactory { HostName = "localhost"};
3. using var connection = factory.CreateConnection();
4. using var channel = connection.CreateModel();
6. //定义接收返回结果的队列
7. var replyQueueName = channel.QueueDeclare().QueueName;
8. var consumer = new EventingBasicConsumer(channel);
10. consumer.Received += (model, ea) =>
11. {
12.     var body = ea.Body.ToArray();
13.     var response = Encoding.UTF8.GetString(body);
14.     Console.WriteLine(" [.] Got '{0}'", response);
15. };
17. //发送消息
18. var correlationId = Guid.NewGuid().ToString(); //消息唯一性
19. var props = channel.CreateBasicProperties();
20. props.CorrelationId = correlationId;
21. props.ReplyTo = replyQueueName; //回调队列名称
23. string message = "30";
24. var messageBytes = Encoding.UTF8.GetBytes(message);
25. channel.BasicPublish(
26.     exchange: "",
27.     routingKey: "rpc_queue",
28.     basicProperties: props,
29.     body: messageBytes);
31. channel.BasicConsume(
32.     consumer: consumer,
33.     queue: replyQueueName,
34.     autoAck: true);
37. //消费者
38. var factory = new ConnectionFactory { HostName = "localhost" };
39. using var connection = factory.CreateConnection();
40. using var channel = connection.CreateModel();
42. channel.QueueDeclare(queue: "rpc_queue",
43.                              durable: false,
44.                              exclusive: false,
45.                              autoDelete: false,
46.                              arguments: null);
47. channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
48. var consumer = new EventingBasicConsumer(channel);
49. channel.BasicConsume(queue: "rpc_queue",
50.                      autoAck: false,
51.                      consumer: consumer);
52. Console.WriteLine(" [x] Awaiting RPC requests");
53. consumer.Received += (model, ea) =>
54. {
55.     string response = string.Empty;
57.     var body = ea.Body.ToArray();
58.     var props = ea.BasicProperties;
59.     var replyProps = channel.CreateBasicProperties();
60.     replyProps.CorrelationId = props.CorrelationId;
62.     try
63.     {
64.         var message = Encoding.UTF8.GetString(body);
65.         int n = int.Parse(message);
66.         Console.WriteLine($" [.] Fib({message})");
67.         response = FibHelper.Fib(n).ToString();
68.     }
69.     catch (Exception e)
70.     {
71.         Console.WriteLine($" [.] {e.Message}");
72.         response = string.Empty;
73.     }
74.     finally
75.     {
76.         //回调到生产者队列
77.         var responseBytes = Encoding.UTF8.GetBytes(response);
78.         channel.BasicPublish(exchange: string.Empty,
79.                              routingKey: props.ReplyTo,
80.                              basicProperties: replyProps,
81.                              body: responseBytes);
82.         channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
83.     }
84. };
87. //执行函数，并返回结果
88. public class FibHelper
89. {
90.      public static int Fib(int n)
91.      {
92.          if (n == 0 || n == 1)
93.             return n;
94.          return Fib(n - 1) + Fib(n - 2);
95.      }
96. }

复制代码

RPC模式不是消息传递模式，消息只会被一个斲丧者斲丧。
7、发布者确认模式

发布者确认模式（Publisher Confirmation）是 RabbitMQ 提供的一种机制，用于确保消息被成功
发送到交换机（exchange）并被接收到，以及确保消息被正确地路由到队列中。在传统的消息发
布过程中，发布者发送消息到交换机后，并不知道消息是否已经被正确地处理惩罚。为相识决这个问
题，RabbitMQ 提供了发布者确认模式，答应发布者确认消息是否已经被成功接收到。

1. //生产者
2. var factory = new ConnectionFactory() { HostName = "localhost" };
3. using (var connection = factory.CreateConnection())
4. using (var channel = connection.CreateModel())
5. {
6.     // 设置信道为确认模式
7.     channel.ConfirmSelect();
9.     // 声明一个队列
10.     channel.QueueDeclare(queue: "hello",
11.                          durable: false,
12.                          exclusive: false,
13.                          autoDelete: false,
14.                          arguments: null);
16.     // 消息内容
17.     string message = "Hello World!";
18.     var body = Encoding.UTF8.GetBytes(message);
20.     try
21.     {
22.         // 发送消息
23.         channel.BasicPublish(exchange: "",
24.                              routingKey: "",
25.                              basicProperties: null,
26.                              body: body);
28.         // 等待消息确认
29.         if (channel.WaitForConfirms())
30.         {
31.             Console.WriteLine(" [x] Sent {0}", message);
32.         }
33.         else
34.         {
35.             Console.WriteLine(" [x] Failed to send {0}", message);
36.         }
37.     }
38.     catch (Exception ex)
39.     {
40.         Console.WriteLine($"An error occurred: {ex.Message}");
41.     }
42. }
44. Console.WriteLine(" Press [enter] to exit.");
45. Console.ReadLine();
47. //消费者
48. var factory = new ConnectionFactory() { HostName = "localhost" };
49. using (var connection = factory.CreateConnection())
50. using (var channel = connection.CreateModel())
51. {
52.     // 声明一个队列
53.     channel.QueueDeclare(queue: "hello",
54.                          durable: false,
55.                          exclusive: false,
56.                          autoDelete: false,
57.                          arguments: null);
59.     // 创建消费者
60.     var consumer = new EventingBasicConsumer(channel);
62.     // 消费消息
63.     consumer.Received += (model, ea) =>
64.     {
65.         var body = ea.Body;
66.         var message = Encoding.UTF8.GetString(body);
67.         Console.WriteLine(" [x] Received {0}", message);
68.     };
70.     // 消费者开始接收消息
71.     channel.BasicConsume(queue: "hello",
72.                          autoAck: true,
73.                          consumer: consumer);
75.     Console.WriteLine(" Press [enter] to exit.");
76.     Console.ReadLine();
77. }

复制代码

五、RabbitMQ相干属性描述

上述代码中，有很多属性的设置，下面解释一下。
名称
值
描述
durable
布尔
队列是否长期化。如果将队列声明为长期化，那么当 RabbitMQ 服务器重启时，队列将被重新声明。长期化队列中的消息会被存储在磁盘上，因此即使在服务器重启后，消息也不会丢失。
exclusive
布尔
是否为排他队列。当队列被声明为排他时，只有声明该队列的毗连（connection）才能使用它。一旦毗连关闭，排他队列就会被删除。这种队列通常是用于临时使命，只答应声明它的毗连使用，不会被其他毗连访问。
autoDelete
布尔
队列是否自动删除。如果将队列声明为自动删除，则在最后一个毗连订阅它的斲丧者取消订阅后，队列将被自动删除。这种属性通常与临时队列一起使用，以确保在不再需要队列时它会被清算。
arguments
类
用于设置队列或交换机的额外参数的选项。这些参数可以用于定制化队列或交换机的举动，以满足特定的需求。比如，设置队列的最大长度、消息在队列中的最大存活时间。
autoAck
布尔
消息是否自动确认。它是指在斲丧者从队列中接收到消息后，是否自动确认消息的斲丧。当设置为 true 时，表示斲丧者会自动确认收到的消息，此时队列中表示该消息已被斲丧成功了。当设置为 false 时，表示斲丧者需要显式地调用确认方法来告知 RabbitMQ 已经成功处理惩罚了消息，否则消息将被重新放回队列，等候其他斲丧者处理惩罚。
basicProperties
类
是指在发布消息时可以携带的消息属性。这些属性包含了有关消息的元数据信息，例如消息的优先级、消息的过期时间、消息的类型等等。

---

总结

RabbitMQ 是一个消息队列，主要作用就是异步、顺序性、削峰等。
七种队列模式，可以根据不同的场景详细使用。
1. 简单队列模式

最简单的消息模式。一个生产者发送消息到一个队列，一个斲丧者从队列中接收消息并处理惩罚。适用
于单个生产者-单个斲丧者的简单场景。
2. 工作队列模式
多个斲丧者共同斲丧消息。斲丧者从队列中取出消息并处理惩罚，消息会均匀地分配给斲丧者。
是基于简单队列模式的缺点，做了提升。适用于负载均衡和使命分发的场景。
3. 发布/订阅模式
生产者将消息发送到交换机，交换机将消息广播到所有与之绑定的队列。多个斲丧者可以订阅不同
的队列，从而接收消息的副本。适用于消息广播和关照的场景。
4. 路由模式
生产者发送消息到交换机，并使用路由键指定消息的目标队列。交换机根据消息的路由键将消息路
由到与之匹配的队列中。适用于根据消息内容举行精确路由的场景。
5. 主题模式
类似于路由模式，但是路由键可以使用通配符举行匹配。适用于消息的多样化路由和灵活的匹配需
求。
6. RPC模式
客户端（RPC请求者）发送请求消息到队列中，并等候服务器（RPC响应者）返反响应消息。
服务器监听请求队列，处理惩罚请求并将响应发送回客户端指定的队列。适用于需要请求-响应式通信
的场景，类似于远程调用。
7. 发布者确认模式

发布者确认模式是 RabbitMQ 提供的一种机制，用于确保消息在发送到交换机并被路由到队列时的
可靠性。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。