.net Core 3.1 使用RabbitMQ(一)
1. 为什么要用消息队列
- 消息队列是啥:消息队列(Message Queue),一种先进先出的队列结构,用于处理Message。
- 消息队列解决的问题:在处理大量并发请求时,同步处理往往会发生阻塞,特别是在更新数据库或者调用其他组件时。消息队列能够实现请求的异步处理、应用解耦、流量削锋和消息通讯,同时能够有效防止阻塞导致的消息丢失。
- 消息队列系统,一般都包含3个角色:队列服务端,队列的生产者,队列的消费者。这种架构与当下流行的分布式架构极为契合。
2. RabbitMQ是啥
- RabbitMQ是使用Erlang编写的一个开源的消息队列,本身支持很多的协议:AMQP,XMPP, SMTP, STOMP,也正因如此,它非常重量级,更适合于企业级的开发。同时实现了Broker构架,这意味着消息在发送给客户端时先在中心队列排队。对路由,负载均衡或者数据持久化都有很好的支持。
- 它支持开放的高级消息队列协议 (AMQP,Advanced Message Queuing Protocol),从根本上避免了生产厂商的封闭,使用任何语言的各种客户都可以从中受益。这种协议提供了相当复杂的消息传输模式,所以基本上不需要MassTransit或NServiceBus的配合。它还具有“企业级”的适应性和稳定性。这些东西对我的客户来说十分的有吸引力。
3. 如何安装RabbitMQ
- 安装RabbitMQ(下载地址: https:// github.com/rabbitmq/rab bitmq-server/releases/tag/v3.7.26 )(安装之前先安装 Erlang)
默认的账户密码都是guest
TIPS:上述操作也可以通过命令行来完成
查看所有用户
rabbitmqctl list_users
添加一个用户
rabbitmqctl add_user admin 123456
配置权限
rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"
查看用户权限
rabbitmqctl list_user_permissions admin
设置tag
rabbitmqctl set_user_tags admin administrator
删除用户(安全起见,删除默认用户)
rabbitmqctl delete_user guest
4. 先了解一下RabbitMQ的队列形式
在使用RabbitMQ之前,咱们先捋一捋RabbitMQ一共有几种使用方式,常用的5中队列
- 第一种:简单队列(一对一模式)耦合性高,生产消费一一对应。
- 第二种:WorkQueues(工作队列),一个生产,多个消费者共同处理消息。
- 第三种:订阅模式,1一个生产者多个消费者,每一个消费者有自己的一个队列,生产者直接将消息发送给交换机,交换机将消息发送给队列,每一个队列都需要绑定到交换机。这种模式可以满足消费者发布一个消息,多个消费者消费同一信息的需求,但C1、C2消费的都是相同的数据,有时我们需要C1和C2消费的信息只有部分差异,比如我们需求:C1消费增加的数据,C2消费编辑、增加和删除的数据。
- 第四种:路由模式,路由模式是在订阅模式基础上的完善,可以在生产消息的时候,加入Key值,与key值匹配的消费者消费信息。但路由模式中,就如三中提到的C1、C2、如果C2对应的类型更多,就需要写很多key值。但其实它只与C1有一点差别。
- 第五种:通配符模式是在路由模式的升级,他允许key模糊匹配。*代表一个词,#代表一个或多个词。通过通配符模式我们就可以将C1对应的一个key准确定为item.add。而C2我们就不需要一一写出key值,而是用item.#代替即可。
5. .net Core3.1如何优雅的使用RabbitMQ
创建生产者
using RabbitMQ.Client;
using System;
using System.Text;
namespace RabbitMQTest
/// <summary>
/// 生产者
/// </summary>
class Program
static void Main(string[] args)
//创建连接工厂
ConnectionFactory factory = new ConnectionFactory
UserName = "admin",//用户名
Password = "lmt456123",//密码
HostName = "localhost"//rabbitmq ip
//创建连接
var connection = factory.CreateConnection();
//创建通道
var channel = connection.CreateModel();
//声明一个队列
channel.QueueDeclare("hello", false, false, false, null);
Console.WriteLine("\nRabbitMQ连接成功,请输入消息,输入exit退出!");
string input;
input = Console.ReadLine();
var sendBytes = Encoding.UTF8.GetBytes(input);
//发布消息
channel.BasicPublish("", "hello", null, sendBytes);
while (input.Trim().ToLower() != "exit");
channel.Close();
connection.Close();
创建一对一消费者
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Text;
namespace RabbitMQConsumer
/// <summary>
/// 消费者
/// </summary>
class Program
static void Main(string[] args)
//创建连接工厂
ConnectionFactory factory = new ConnectionFactory
UserName = "admin",//用户名
Password = "lmt456123",//密码
HostName = "localhost"//rabbitmq ip
//创建连接
var connection = factory.CreateConnection();
//创建通道
var channel = connection.CreateModel();
//事件基本消费者
EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
//接收到消息事件
consumer.Received += (ch, ea) =>
string message = Encoding.Default.GetString(ea.Body.ToArray());
Console.WriteLine($"收到消息: {message}");
// Console.WriteLine($"收到该消息[{ea.DeliveryTag}] 延迟10s发送回执");
// Thread.Sleep(10000);
//确认该消息已被消费
channel.BasicAck(ea.DeliveryTag, false);
// Console.WriteLine($"已发送回执[{ea.DeliveryTag}]");
//启动消费者 设置为手动应答消息
channel.BasicConsume("hello", false, consumer);
Console.WriteLine("消费者已启动");
Console.ReadKey();