Consuming RabbitMQ Messages In ASP.NET Core

Background tasks play a very important role when we are building a distributed system. The most common scenario is consuming the service bus's message. In this article, I'd like to present how to consume the RabbitMQ message via BackgroundService in ASP.NET Core.

Run RabbitMQ Host

We should set up an instance of RabbitMQ. The fastest way is to use Docker.

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

After running the Docker container, we are able to view the management page via http://localhost:15672.

Setup the background service

Here, we create a new class named ConsumeRabbitMQHostedService that is inherited from BackgroundService.

BackgroundService is a base class for implementing a long-running IHostedService. It provides the main work needed to set up the background task.

Here is an example to demonstrate how to consume RabbitMQ messages.

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();
    }
}

Configure Services

We should configure this hosted service with the background task logic in the ConfigureServices method.

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

Result

After running this app, we may get the following output in the terminal.

Turning to the Management UI of RabbitMQ, we find that it creates a new exchange and a new queue.

The next time we try to publish a message to show the background task is running well, we get the following result.

All things went well!

Summary

This article showed you how to consume RabbitMQ messages via BackgroundService in ASP.NET Core. I hope this will help you!