在 ASP.NET Core 中使用 RabbitMQ 消息

在我们构建分布式系统时，后台任务扮演着非常重要的角色。最常见的场景是使用服务总线的消息。在本文中，我想介绍如何在 ASP.NET Core 中通过 BackgroundService 使用 RabbitMQ 消息。

运行 RabbitMQ 主机

我们应该建立一个 RabbitMQ 的实例。最快的方法是使用 Docker。

docker run -p 5672:5672 -p 15672:15672 rabbitmq:management   

运行 Docker 容器后，我们可以通过 http://localhost:15672 查看管理页面。

设置后台服务

在这里，我们创建了一个名为ConsumeRabbitMQHostedService的新类 ，它继承自BackgroundService。

BackgroundService 是用于实现长时间运行的 IHostedService 的基类。它提供了设置后台任务所需的主要工作。

这是一个示例来演示如何使用 RabbitMQ 消息。

public class ConsumeRabbitMQHostedService : BackgroundService  
{  
    private readonly ILogger _logger;  
    private IConnection _connection;  
    private IModel _channel;  
  
    public ConsumeRabbitMQHostedService(ILoggerFactory loggerFactory)  
    {  
        this._logger = loggerFactory.CreateLogger<ConsumeRabbitMQHostedService>();  
        InitRabbitMQ();  
    }  
  
    private void InitRabbitMQ()  
    {  
        var factory = new ConnectionFactory { HostName = "localhost" };  
  
        // create connection  
        _connection = factory.CreateConnection();  
  
        // create channel  
        _channel = _connection.CreateModel();  
  
        _channel.ExchangeDeclare("demo.exchange", ExchangeType.Topic);  
        _channel.QueueDeclare("demo.queue.log", false, false, false, null);  
        _channel.QueueBind("demo.queue.log", "demo.exchange", "demo.queue.*", null);  
        _channel.BasicQos(0, 1, false);  
  
        _connection.ConnectionShutdown += RabbitMQ_ConnectionShutdown;  
    }  
  
    protected override Task ExecuteAsync(CancellationToken stoppingToken)  
    {  
        stoppingToken.ThrowIfCancellationRequested();  
  
        var consumer = new EventingBasicConsumer(_channel);  
        consumer.Received += (ch, ea) =>  
        {  
            // received message  
            var content = System.Text.Encoding.UTF8.GetString(ea.Body);  
  
            // handle the received message  
            HandleMessage(content);  
            _channel.BasicAck(ea.DeliveryTag, false);  
        };  
  
        consumer.Shutdown += OnConsumerShutdown;  
        consumer.Registered += OnConsumerRegistered;  
        consumer.Unregistered += OnConsumerUnregistered;  
        consumer.ConsumerCancelled += OnConsumerConsumerCancelled;  
  
        _channel.BasicConsume("demo.queue.log", false, consumer);  
        return Task.CompletedTask;  
    }  
  
    private void HandleMessage(string content)  
    {  
        // we just print this message   
        _logger.LogInformation($"consumer received {content}");  
    }  
      
    private void OnConsumerConsumerCancelled(object sender, ConsumerEventArgs e)  {  }  
    private void OnConsumerUnregistered(object sender, ConsumerEventArgs e) {  }  
    private void OnConsumerRegistered(object sender, ConsumerEventArgs e) {  }  
    private void OnConsumerShutdown(object sender, ShutdownEventArgs e) {  }  
    private void RabbitMQ_ConnectionShutdown(object sender, ShutdownEventArgs e)  {  }  
  
    public override void Dispose()  
    {  
        _channel.Close();  
        _connection.Close();  
        base.Dispose();  
    }  
}  

配置服务

我们应该使用 ConfigureServices 方法中的后台任务逻辑来配置这个托管服务。

public void ConfigureServices(IServiceCollection services)  
{  
    // others ...  
      
    services.AddHostedService<ConsumeRabbitMQHostedService>();  
}  

结果

运行此应用程序后，我们可能会在终端中得到以下输出。

下次我们尝试发布消息以显示后台任务运行良好时，我们得到以下结果。

一切顺利！

本文向您展示了如何在 ASP.NET Core 中通过 BackgroundService 使用 RabbitMQ 消息。我希望这能帮到您！