<< European ASP.NET Core Hosting :: Using Docker to Store ASP.NET Core Kestrel Certificates | European ASP.NET Core Hosting :: How to Use Auth Cookies in ASP.NET Core >>

European ASP.NET Core Hosting :: Consuming RabbitMQ Messages In ASP.NET Core

March 26, 2019 11:42 by author

Peter

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.

Consuming RabbitMQ Messages In ASP.NET Core
Setup the BackgroundService
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 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.