在 .NET Core 8 中实现 RabbitMQ

如果您喜欢此文章，请收藏、点赞、评论，谢谢，祝您快乐每一天。

RabbitMQ 是一个功能强大且广泛使用的开源消息代理，用于促进分布式系统之间的通信。它实现了高级消息队列协议 (AMQP)，使应用程序能够通过队列发送和接收消息，从而确保可靠高效的数据传输。RabbitMQ 支持各种消息传递模式，例如发布/订阅、请求/回复和点对点，使其适用于众多用例。

在 .NET 8 环境中，集成 RabbitMQ 可以显著增强应用程序的可扩展性和弹性。借助最新的 .NET 8 功能，开发人员可以在实施 RabbitMQ 时享受到提升的性能和安全性。本文将指导您设置 RabbitMQ，在 .NET 8 环境中进行配置，并演示发送和接收消息的实际示例。学完本教程后，您将深入了解如何利用 RabbitMQ 使用 .NET 8 构建可扩展的分布式系统。

在本文中，我们可以看到如何在 .NET Core 8 中实现 RabbitMQ。

要在 .NET 8 应用程序中实现 RabbitMQ，您需要使用特定的 NuGet 包来促进与 RabbitMQ 的通信。您应该包含的两个主要包是 MassTransit 和 MassTransit.RabbitMQ。MassTransit 是一种广泛使用的 .NET 服务总线，可简化基于消息的应用程序开发，而 MassTransit.RabbitMQ 则提供与 RabbitMQ 的必要集成。

以下是将这些包添加到 .NET 8 项目的方法：

dotnet add package MassTransit

2.安装MassTransit.RabbitMQ：

dotnet add package MassTransit.RabbitMQ

然后在 appsetting.json 文件中添加此 RabbitMQ 配置代码：

"RabbitMQ": {
"HostName": "localhost",
"UserName": "guest",
"Password": "guest"

}

现在添加一个用于 RabbitMQ 相关设置配置的类

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace SalesPulse.MessageBroker
{
public class RabbitMQSetting
{
public string? HostName { get; set; }
public string? UserName { get; set; }
public string? Password { get; set; }
}

//RabbitMQ Queue name
public static class RabbitMQQueues
{
public const string OrderValidationQueue = "orderValidationQueue";
public const string AnotherQueue = "anotherQueue";
public const string ThirdQueue = "thirdQueue";
}
}

我们需要一个发布者服务类来发布消息。为了提高代码的可重用性，最好创建一个通用的服务类。消息发布者服务类如下：

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Microsoft.Extensions.Options;
using Newtonsoft.Json;
using RabbitMQ.Client;

namespace SalesPulse.MessageBroker.Services
{
public class RabbitMQPublisher<T> : IRabbitMQPublisher<T>
{
private readonly RabbitMQSetting _rabbitMqSetting;

public RabbitMQPublisher(IOptions<RabbitMQSetting> rabbitMqSetting)
{
_rabbitMqSetting = rabbitMqSetting.Value;
}

public async Task PublishMessageAsync(T message, string queueName)
{

var factory = new ConnectionFactory
{
HostName = _rabbitMqSetting.HostName,
UserName = _rabbitMqSetting.UserName,
Password = _rabbitMqSetting.Password
};

using var connection = factory.CreateConnection();
using var channel = connection.CreateModel();
channel.QueueDeclare(queue: queueName, durable: false, exclusive: false, autoDelete: false, arguments: null);

var messageJson = JsonConvert.SerializeObject(message);
var body = Encoding.UTF8.GetBytes(messageJson);

await Task.Run(() => channel.BasicPublish(exchange: "", routingKey: queueName, basicProperties: null, body: body));
}
}
}

并创建一个接口类

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace SalesPulse.MessageBroker.Services
{
public interface IRabbitMQPublisher<T>
{
Task PublishMessageAsync(T message, string queueName);
}
}

因此，我们的发布者服务类已经准备好了，现在我们需要一个发布者模型类，该模型类只不过是我们的发布者消息数据的实体。

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace SalesPulse.MessageBroker
{
public class OrderValidation
{
public Guid? OrderEntryById { get; set; }
public string? OrderEntryByCode { get; set; }
public List<ProductValidation>? Products { get; set; }
public string? SupervisorDeviceId { get; set; }
public long OrderNumber { get; set; }
public Guid NotificationReceiverId { get; set; }
}

public class ProductValidation
{
public Guid? ProductId { get; set; }
public string? ProductCode { get; set; }
public decimal? OrderAmount { get; set; }
public string? ProductName { get; set; }
}

}

发布端的工作基本完成了，现在我们需要在控制器构造函数中注入我们的发布服务类，并使用发布服务方法发布消息。在我的例子中，订单录入时，我需要验证订单，并根据验证结果，让一定数量的用户收到推送通知。因此，我将消息发布到订单验证队列中。

// Prepare validation data

var orderValidation = await _chemistService.GetOrderValidationInfoByOrderEntry(order.OrderEntryBy, order, orderDetailsList);

// publish order validation data
await _orderValidationMqPublisher.PublishMessageAsync(orderValidation, RabbitMQQueues.OrderValidationQueue);

最后在 program.cs 文件中配置这些服务。使用 AddScoped 来设置 Publisher 服务类的依赖生命周期。

builder.Services.Configure<RabbitMQSetting>(configuration.GetSection("RabbitMQ"));
builder.Services.AddScoped(typeof(IRabbitMQPublisher<>), typeof(RabbitMQPublisher<>));

发布端的工作已经完成了。现在我们需要配置消费者端……让我们看看如何配置……

在appsetting等设置类中的设置过程与producer相同。并且在消费者端添加相同的发布者类，或者您必须在两个类之间进行映射才能获取数据。

这里我需要一个消费者服务类，它基本上是使用来自队列的消息。

using System;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using Newtonsoft.Json;
using PulseWorker.Data;
using PulseWorker.Domain;
using PulseWorker.Service;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace PulseWorker.MessageBroker
{
public class OrderValidationMessageConsumerService : BackgroundService
{
private readonly IServiceProvider _serviceProvider;
private readonly ILogger<OrderValidationMessageConsumerService> _logger;
private readonly RabbitMQSetting _rabbitMqSetting;
private IConnection _connection;
private IModel _channel;

public OrderValidationMessageConsumerService(IOptions<RabbitMQSetting> rabbitMqSetting, IServiceProvider serviceProvider, ILogger<OrderValidationMessageConsumerService> logger)
{
_rabbitMqSetting = rabbitMqSetting.Value;
_serviceProvider = serviceProvider;
_logger = logger;

var factory = new ConnectionFactory
{
HostName = _rabbitMqSetting.HostName,
UserName = _rabbitMqSetting.UserName,
Password = _rabbitMqSetting.Password
};
_connection = factory.CreateConnection();
_channel = _connection.CreateModel();
}

protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
StartConsuming(RabbitMQQueues.OrderValidationQueue, stoppingToken);
await Task.CompletedTask;
}

private void StartConsuming(string queueName, CancellationToken cancellationToken)
{
_channel.QueueDeclare(queue: queueName, durable: false, exclusive: false, autoDelete: false, arguments: null);

var consumer = new EventingBasicConsumer(_channel);
consumer.Received += async (model, ea) =>
{
var body = ea.Body.ToArray();
var message = Encoding.UTF8.GetString(body);

bool processedSuccessfully = false;
try
{
processedSuccessfully = await ProcessMessageAsync(message);
}
catch (Exception ex)
{
_logger.LogError($"Exception occurred while processing message from queue {queueName}: {ex}");
}

if (processedSuccessfully)
{
_channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
}
else
{
_channel.BasicReject(deliveryTag: ea.DeliveryTag, requeue: true);
}
};

_channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer);
}

private async Task<bool> ProcessMessageAsync(string message)
{
try
{
using (var scope = _serviceProvider.CreateScope())
{
var notificationService = scope.ServiceProvider.GetRequiredService<INotificationService>();
var notificationRepository = scope.ServiceProvider.GetRequiredService<INotificationRepository>();
var orderValidationRepository = scope.ServiceProvider.GetRequiredService<IValidationRepository>();
var orderValidation = JsonConvert.DeserializeObject<OrderValidation>(message);

if (string.IsNullOrEmpty(orderValidation?.SupervisorDeviceId) || orderValidation.Products == null)
{
return true;
}

foreach (var item in orderValidation.Products)
{
bool isExceed = await orderValidationRepository.IsValidationAmountExceed(item.ProductId);

if (isExceed)
{
var notificationModel = new NotificationModel
{
DeviceId = orderValidation.SupervisorDeviceId,
IsAndroiodDevice = true,
Title = "OrderValidation",
Body = $"Chemist Order – {orderValidation.OrderNumber} exceeds the validation amount for product {item.ProductName}-{item.ProductCode}"
};

var notificationResponse = await notificationService.SendNotificationAsync(notificationModel);

if (!notificationResponse.IsSuccess)
{
return false;
}

var notification = new Notification
{
Title = notificationModel.Title,
Body = notificationModel.Body,
ReceiverId = orderValidation.NotificationReceiverId,
TypeIdentifyId = orderValidation.OrderNumber.ToString(),
Type = notificationModel.Title,
};

await notificationRepository.AddNotificationAsync(notification);
await notificationRepository.SaveNotificationAsync();

return true;
}
}

return true;
}
}
catch (Exception ex)
{
_logger.LogError($"Error processing message: {ex.Message}");
return false;
}
}

public override void Dispose()
{
_channel.Close();
_connection.Close();
base.Dispose();
}
}
}

在这里，我的消费者服务是消费消息并验证订单，如果验证规则中断，则发送推送通知。

现在在 program.cs 中配置服务类托管服务

// RabbitMQ Configuration
builder.Services.Configure<RabbitMQSetting>(builder.Configuration.GetSection("RabbitMQ"));
// Register the consumer service as a hosted service only
builder.Services.AddHostedService<OrderValidationMessageConsumerService>();

就是这样，这是 .NET 8 中 RabbitMQ 非常基本和简单的设置。

希望本文有助于在 .NET Core API 中实现 RabbitMQ。

如果您喜欢此文章，请收藏、点赞、评论，谢谢，祝您快乐每一天。