VEC系列-RabbitMQ 入门条记

消息队列（MQ）对于开发者来说是一个经常听到的词汇，但在现实开发中，大多数人并不会真正用到它。网上已经有很多关于 MQ 概述和原理的具体讲解，官网文档和技能博客也都介绍得很深入，因此，我在这里就不再赘述。
  我一直认为，学习一项技能不仅要知道它是什么，更告急的是知道怎么用，以及在哪些场景下应该用。所以这篇文章重要就是站在一个新手的角度进行形貌以及实现MQ的现实运用。
  使用MQ的常见情景

• 系统解耦：好比电商系统，订单系统 → 库存系统 → 物流系统 订单系统发送“新订单”消息到 MQ，库存系统和物流系统各自订阅处理。纵然库存系统或物流系统短暂不可用，消息仍然可以暂存，系统团体不会受影响。这一方面说实话不是架构师也没须要太过关注，究竟系统的底层普通开发也没这个资格去搭建。只是用于了解，不要因为这段话拦阻学习的脚步。
  
• 流量削峰，降低并发：这个比较好理解，也是最能遇到的环境。用户哀求先进入 MQ 队列，由背景的消耗端按照数据库的最大承载能力逐步处理哀求。确保数据库不会被瞬间压垮，进步系统稳定性。还是电商系统常用些。
  
• 异步任务处理：邮件、短信、推送关照，日志处理等。
  

理论上MQ能做的不止这些，抛砖引玉，一起深入学习吧。
对MQ进行拆分理解

MQ里常说生产者，消耗者等。我会通过简单的例子来形貌：

• 生产者：一个游戏，我是GM，我要发送公告，玩家分为普通玩家和VIP玩家等。在这里，发布公告的人就是消息的生产者。应该很好理解嗷？
  
• 交换器：如上述，有普通玩家和VIP玩家等，我的公告在普通玩家面前必然是拽的很啊，但是VIP玩家面前还是要舔下的……那么我会发布一条给普通玩家的消息，和一条给VIP玩家的消息。交换器的作用在我看来就是消息的承载体，类似一条运输船，负责把消息运输给玩家们。产生消息的地方很多，但是交换器不用关心是谁发布了消息，他只承载你的消息。
  
• 队列：如上述，有了运输船。那么队列有点像是码头了。普通玩家进普通码头，VIP玩家进黄金码头。各自码头停泊各自的船。总不会在普通码头取出黄金码头的货哦？
  

   补充：交换器是有类型的：Direct（直连交换器）Fanout（扇形交换器）Topic（主题交换器）Headers（头交换器）
  概念不多说了。比较常用的是Direct，Fanout
  Direct：通过路由键进行匹配，运输船是一艘，但是分为普通区和VIP区，玩家凭借船票（路由键）进行取货（取消息）
  Fanout：只要是是绑定了某个交换器的队列都能进行取货。玩家进普通码头就拿普通货，进黄金码头拿黄金货。固然这是举例子，玩家的队列还是要看你如何分配的。
  

• 消耗者：说了这么多，玩家就是消耗者嗷。
  

MQ代码演示 

   最新代码是通过 事件总线 来跨方法通报信息和触发动作。通过发布和订阅事件，模块之间可以或许解耦通信，使得事件的发布和处理不再依赖于直接调用方法的方式，而是通过事件总线进行跨模块、跨方法的异步通报。这种方式进步了系统的灵活性和扩展性，同时保持了模块之间的松耦合。
  长代码警告，有兴趣可以fork仓库进行现实操练 VerEasy.Core。
  须要的知识点大致云云，通过代码+解释的形式来演示更好理解。
我这里是NETCore项目，所以还是接口的形式方便依赖注入。
接口部分代码

1.     public interface IRabbitMQPersistentConnection
2.     {
3.         /// <summary>
4.         /// 是否已经连接：判断MQ是否是连接状态
5.         /// </summary>
6.         bool IsConnected { get; }
8.         /// <summary>
9.         /// 尝试连接：断连重连方法
10.         /// </summary>
11.         /// <returns></returns>
12.         Task<bool> TryConnectAsync();
14.         /// <summary>
15.         /// 唯一通道：发布通道可以随时关闭，消费通道需要保持打开状态，否则无法进行消费。
16.         /// </summary>
17.         IChannel Channel { get; }
19.         /// <summary>
20.         /// 唯一连接：同理，一个连接可以有N个通道，无需建立过多连接。
21.         /// </summary>
22.         IConnection Connection { get; }
24.         /// <summary>
25.         /// 释放
26.         /// </summary>
27.         /// <returns></returns>
28.         Task DisposeAsync();
30.         /// <summary>
31.         /// 发布：发布消息
32.         /// </summary>
33.         /// <param name="msg"></param>
34.         /// <param name="exChangeName"></param>
35.         /// <param name="routeKey"></param>
36.         /// <param name="type"></param>
37.         /// <returns></returns>
38.         Task PublishAsync(string msg, string exChangeName = "VECLOG", string routeKey = "", string type = ExchangeType.Fanout);
40.         /// <summary>
41.         /// 订阅：订阅队列。
42.         /// </summary>
43.         /// <returns></returns>
44.         Task SubscribeAsync(string exChangeName = "VECLOG", string routeKey = "", string type = ExchangeType.Fanout);
45.     }

复制代码

 接口实现部分代码

1.     public class RabbitMQPersistentConnection : IRabbitMQPersistentConnection
2.     {
3.         //构造函数注入,获取MQ的地址账号密码端口,如果不传就用我默认配置的。
4.         public RabbitMQPersistentConnection(IConnectionFactory? connectionFactory = null, int retryCount = 5)
5.         {
6.             _connectionFactory = connectionFactory ?? new ConnectionFactory
7.             {
8.                 HostName = Appsettings.AppStr("RabbitMQ:Connection:HostName"),
9.                 UserName = Appsettings.AppStr("RabbitMQ:Connection:UserName"),
10.                 Password = Appsettings.AppStr("RabbitMQ:Connection:PassWord"),
11.                 Port = Appsettings.AppStr("RabbitMQ:Connection:Port").ObjToInt()
12.             };
13.             //使用Policy进行重连，这个是重连次数=5
14.             _retryCount = retryCount;
15.         }
16.         //私有变量，获取连接成功时创建的Mq通道。
17.         private IChannel _channel = default!;
19.         public IChannel Channel
20.         {
21.             get
22.             {
23.                 return _channel;
24.             }
25.         }
27.         /// <summary>
28.         /// RabbitMQ 连接工厂
29.         /// </summary>
30.         private readonly IConnectionFactory _connectionFactory;
32.         /// <summary>
33.         /// 私有变量 RabbitMQ 连接上下文
34.         /// </summary>
35.         private IConnection _connection = default!;
37.         /// <summary>
38.         /// 重连次数
39.         /// </summary>
40.         private readonly int _retryCount;
42.         /// <summary>
43.         /// 标志是否已释放
44.         /// </summary>
45.         private bool _disposed;
47.         /// <summary>
48.         /// 是否有效连接
49.         /// </summary>
50.         public bool IsConnected
51.         {
52.             get
53.             {
54.                 return _connection != null && _connection.IsOpen && !_disposed;
55.             }
56.         }
58.         public IConnection Connection
59.         {
60.             get
61.             {
62.                 return _connection;
63.             }
64.         }
66.         /// <summary>
67.         /// 手动释放
68.         /// </summary>
69.         /// <returns></returns>
70.         public async Task DisposeAsync()
71.         {
72.             if (_disposed) return;
74.             _disposed = true;
76.             try
77.             {
78.                 await _connection.DisposeAsync();
79.             }
80.             catch (IOException ex)
81.             {
82.                 Console.WriteLine(ex.Message);
83.             }
84.         }
86.         /// <summary>
87.         /// 重连机制
88.         /// </summary>
89.         /// <returns></returns>
90.         public async Task<bool> TryConnectAsync()
91.         {
92.             var policy = Policy.Handle<SocketException>()//捕获连接异常
93.                 .Or<BrokerUnreachableException>()//无法连接异常
94.                 .WaitAndRetryAsync(_retryCount, x =>
95.                 TimeSpan.FromSeconds(Math.Pow(2, x)), (ex, time) =>
96.                 {
97.                     //日志
98.                 });
100.             try
101.             {
102.                 await policy.ExecuteAsync(async () =>
103.                 {
104.                     //重建连接【赋值给私有化变量,通过get同步给接口里的Connection和Channel】
105.                     _connection = await _connectionFactory.CreateConnectionAsync();
106.                     _channel = await _connection.CreateChannelAsync();
107.                 });
109.                 //如果连接成功
110.                 if (IsConnected)
111.                 {
112.                     // 连接成功后，注册连接关闭、异常、阻塞的事件处理程序
113.                     _connection.ConnectionShutdownAsync += OnConnectionShutdownAsync;
114.                     _connection.CallbackExceptionAsync += OnCallbackExceptionAsync;
115.                     _connection.ConnectionBlockedAsync += OnConnectionBlockedAsync;
116.                     return true;
117.                 }
118.                 else
119.                 {
120.                     return false;
121.                 }
122.             }
123.             catch (Exception ex)
124.             {
125.                 Console.WriteLine($"重连失败，最终抛出异常: {ex.Message}");
126.                 return false;
127.             }
128.         }
130.         private async Task OnConnectionShutdownAsync(object sender, ShutdownEventArgs e)
131.         {
132.             if (_disposed) return;
133.             Console.WriteLine("RabbitMQ连接关闭，正在尝试重连...");
134.             await TryConnectAsync();
135.         }
137.         private async Task OnCallbackExceptionAsync(object sender, CallbackExceptionEventArgs e)
138.         {
139.             if (_disposed) return;
140.             Console.WriteLine($"RabbitMQ连接出现异常，正在尝试重连... 异常信息: {e.Exception.Message}");
141.             await TryConnectAsync();
142.         }
144.         private async Task OnConnectionBlockedAsync(object sender, ConnectionBlockedEventArgs e)
145.         {
146.             if (_disposed) return;
147.             Console.WriteLine("RabbitMQ连接被阻塞，正在尝试重连...");
148.             await TryConnectAsync();
149.         }
151.         //发布消息
152.         public async Task PublishAsync(string msg, string exChangeName = "VECLOG", string routeKey = "", string type = ExchangeType.Fanout)
153.         {
154.             //判断是否连接状态，没有连接就重连
155.             if (!IsConnected)
156.             {
157.                 await TryConnectAsync();
158.             }
160.             //创建通道，因为是发布消息，通道不用常打开，所以使用了USING
161.             using var channel = await _connection.CreateChannelAsync();
162.             //【ExchangeDeclareAsync】声明交换机，exchange：交换机名称，type：交换机类型
163.             await channel.ExchangeDeclareAsync(exchange: exChangeName, type: type);
164.             
165.             //msg就是消息,需要传递Byte[]
166.             var body = Encoding.UTF8.GetBytes(msg);
167.             
168.             //启动消息持久化,我的项目里使用MQ来进行公告的推送,使用的Fanout类型交换机，故此消息保持持久化。
169.             var properties = new BasicProperties()
170.             {
171.                 Persistent = true,
172.             };
174.             //发布消息
175.             await channel.BasicPublishAsync(
176.                 exchange: exChangeName,
177.                 routingKey: routeKey,
178.                 mandatory: false,
179.                 basicProperties: properties,
180.                 body: body);
181.         }
183.         //订阅消息
184.         public async Task SubscribeAsync(string exChangeName = "VECLOG", string routeKey = "", string type = ExchangeType.Fanout)
185.         {
186.             if (!IsConnected)
187.             {
188.                 await TryConnectAsync();
189.             }
190.             
191.             //【queue】队列
192.             string queueName = string.IsNullOrWhiteSpace(routeKey) ? exChangeName : routeKey;
193.             //【durable】持久化队列,MQ服务器不会删除它。
194.             QueueDeclareOk queueDeclareResult = await Channel.QueueDeclareAsync(
195.                  queue: queueName,
196.                   durable: true,
197.                   exclusive: false,
198.                   autoDelete: false);
200.             //根据queue,exchange,routingKey 对 交换机和队列进行绑定，如果是Fanout类型不需要routeKey。
201.             await Channel.QueueBindAsync(queue: queueName, exchange: exChangeName, routingKey: routeKey);
202.             //创建消费者
203.             var consumer = new AsyncEventingBasicConsumer(Channel);
204.             //消费者消费后执行方法
205.             consumer.ReceivedAsync += async (model, ea) =>
206.             {
207.                 byte[] body = ea.Body.ToArray();
208.                 var message = Encoding.UTF8.GetString(body);
209.                 //确认消息已被消费,这样后续该消息就不会被该队列继续消费到了。
210.                 await Channel.BasicAckAsync(ea.DeliveryTag, multiple: false);
211.             };
212.             //启动消费者队列，将消费者和队列绑定
213.             await Channel.BasicConsumeAsync(queueName, autoAck: false, consumer: consumer);
214.         }
215.     }

复制代码

 
MQ服务注入

1.             if (Appsettings.AppStr("RabbitMQ:Enable").ObjToBool())
2.             {
3.                 services.AddSingleton<IRabbitMQPersistentConnection>(x =>
4.                 {
5.                     var connectionFactory = new ConnectionFactory()
6.                     {
7.                         HostName = Appsettings.AppStr("RabbitMQ:Connection:HostName"),
8.                         UserName = Appsettings.AppStr("RabbitMQ:Connection:UserName"),
9.                         Password = Appsettings.AppStr("RabbitMQ:Connection:PassWord"),
10.                         Port = Appsettings.AppStr("RabbitMQ:Connection:Port").ObjToInt()
11.                     };
12.                     var mq = new RabbitMQPersistentConnection(connectionFactory);
14.                     return mq;
15.                 });
16.             }

复制代码

 

---

我在注入各种服务时，添加了一些日志进行输出，效果如下：

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