RabbitMQ and MassTransit in .NET Core : Practical Guide

As more and more applications and systems must be highly available and scalable, choosing the right software architecture to build them becomes critically important. In today’s ever-evolving and rapidly changing technological landscape, where demands for efficiency, speed, and reliability continue to grow, architects and developers face the ongoing challenge of creating resilient solutions that can seamlessly adapt to shifting requirements.

To meet these challenges, modern applications and services are increasingly designed as interconnected services and components that communicate with each other to accomplish specific tasks. In this dynamic and fast-paced environment, selecting the right messaging infrastructure and tools is essential for ensuring smooth communication and efficient coordination between these distributed or separate services.

In this blog post, we will explore two powerful tools that facilitate such communication in .NET Core applications: RabbitMQ and MassTransit. By understanding their capabilities, differences, and best-use scenarios, you can make informed decisions about which tool best suits your application’s needs.

RabbitMQ is an open-source messaging and streaming broker software that originally implemented the Advanced Message Queuing Protocol (AMQP). Since its inception, it has been extended with a plugin architecture to support other protocols. Essentially, RabbitMQ is a software where queues are defined so that applications can publish messages to them or consume messages from them.

MassTransit is an open-source messaging library for .NET. It offers a lightweight yet feature-rich framework for implementing various communication patterns such as publish-subscribe (PubSub), request-response, and event-driven architectures. The primary objective of MassTransit is to provide a consistent abstraction layer over message transport, whether it is RabbitMQ or other messaging platforms.

In the world of microservices and distributed architecture, decoupling services from one another stands out as a fundamental principle. By reducing interdependencies, services become less affected by disruptions caused by the availability or failure of other services. Message queuing stands out as a key player in achieving this decoupling. Services can publish messages to a queue, allowing other services to consume them and process them asynchronously. This asynchronous communication model not only enhances system resilience and responsiveness but also improves scalability and agility.

Moreover, in cases when immediate response from other services is not feasible, message queues play a central role in the interaction between services in distributed systems.

Furthermore, message queuing facilitates the implementation of reliable communication patterns such as publish-subscribe and request-response, enabling services to interact seamlessly while maintaining loose coupling. This loose coupling not only enhances fault tolerance but also fosters the evolution and iteration of individual services without disrupting the entire system.

Before we dive into implementations and code, let’s take a few minutes for a quick overview of MassTransit. If you are familiar with MassTransit already, feel free to skip this section.

MassTransit’s architecture revolves around several key building blocks that streamline the implementation of messaging patterns within .NET Core applications. These building blocks provide a cohesive framework for defining message endpoints, handling message consumption, and orchestrating communication flows between services. Let’s explore some of the main components that form the backbone of MassTransit’s functionality:

• Messages
  Messages service the the primary means of communication between services in MassTransit. They contain data and commands exchanges between endpoints. They are strongly typed contracts that be defined used a record, class or an interface. There are two main message types: Events and Commands. A command tells the service to do something whereas an Event signals that something has happened.
• Consumers
  Consumers are responsible for processing incoming messages from message queues or topics. In MassTransit, consumers implement message handlers that specify how messages of a particular type should be processed.
• Producers
  Producers are responsible for sending messages to message queues or topics for consumption by other services. Producers can produce messages using two different ways: Publishing a message or Sending a message. The difference between the two is that when sending a message, the message is delivered to a specific endpoint using a specific address, whereas when publishing a message, the message is broadcasted to all consumers that subscribed to the specific message type.
• Requests
  Request/response is a prevalent messaging pattern in distributed systems, where one service sends a request to another and awaits a response before proceeding. While this approach can introduce latency, especially when services are hosted across different processes or machines, it remains a necessary and often preferred method for certain scenarios. In MassTransit, developers leverage a request client to initiate requests asynchronously and handle responses without blocking the application’s execution. This asynchronous nature, coupled with support for the ‘await’ keyword, ensures that system performance is maintained while waiting for responses from remote services, contributing to a more efficient and responsive distributed architecture.
• Exceptions Exception handling is a critical aspect of building resilient distributed systems with MassTransit. In the event of errors during message processing, MassTransit provides robust mechanisms for managing exceptions. One such mechanism involves routing failed messages to an error queue, where they can be logged and analyzed for troubleshooting purposes.

For the full documentation on MassTransit, please make sure to visit the official documentation available online.

To demonstrate how to work with MassTransit and RabbitMQ, I’ve created a sample project that simulates collecting measurement data from sensors. Occasionally, one of these measurements will indicate an anomaly.

In this simulation, the sensors publish their readings (measurements) to a RabbitMQ queue. On the server side, the TelemetryService consumes these messages. Its primary responsibilities are to validate and process the incoming data. If an anomaly is detected, the TelemetryService publishes an anomaly message to another service, the AnomalyHandlingService, which is responsible for processing and managing the anomalies.

Before we get started, let’s set up the environment with RabbitMQ and MassTransit.

Since the focus of this blog post is on working with RabbitMQ and MassTransit, we will keep the setup of RabbitMQ simple and basic, using a single node. however, the implementation logic is the same for other clustered RabbitMQ configurations.

Since we will be using RabbitMQ as a docker image, use the following command to get started:

docker run -d --hostname rmq-host --name rmq-c -p 15672:15672 -p 5672:5672 rabbitmq:3-management

This will start RabbitMQ in a docker container with the management plugin, and exposes ports 5672 and 15672.

After creating a new project, the next step is to configure MassTransit to work with RabbitMQ. This configuration is typically done in the Program.cs file. For simplicity, we’ll use basic authentication to connect to RabbitMQ.

Adding the necessary setup involves initializing MassTransit, specifying RabbitMQ as the message broker, and defining connection details such as the host, username, and password. This step ensures that your application can publish and consume messages reliably through RabbitMQ.

// Add MassTransit services
builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<TelemetryDataConsumer>();
    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("localhost", "/", host =>
        {
            host.Username("guest");
            host.Password("guest");
        });

        cfg.ConfigureEndpoints(context);
    });
});

MassTransit provides an intuitive way to publish and consume messages from a message broker (in this case, RabbitMQ). To get started, we need to define a Message class, which represents the data structure for the information you want to send to the queue. This class acts as the contract between the publisher and the consumer, ensuring consistency in the data format.

Once the message class is created, we can publish messages to the queue using MassTransit’s API. Later, these messages can be consumed by the appropriate subscribers.

The message class in our example looks like this:

public class TelemetryDataMessage
{
    // Unique identifier for the device sending the telemetry data
    public string DeviceId { get; set; }
    // Timestamp of when the telemetry data was recorded
    public DateTime Timestamp { get; set; }
    public WaterMeasurementData WaterMeasurementData { get; set; }
    //Include metadata about the message, such as data quality or source
    public string DataQuality { get; set; }

    public override string ToString()
    {
        return $"DeviceId: {DeviceId}, Timestamp: {Timestamp}, WaterMeasurementData: {WaterMeasurementData}";
    }
}

In our sample project, the DataInjectorSim is a straightforward console application designed to publish sensor readings to RabbitMQ. To achieve this, we first need to set up MassTransit, as explained earlier in the “Setting Up MassTransit” section.

Once the setup is complete, the application uses a worker class to continuously publish messages to the RabbitMQ queue. This worker simulates the behavior of sensors sending measurement data at regular intervals, providing a steady stream of messages for our system to process.

Publishing a message is done through the Bus, a core component of MassTransit. The Bus encapsulates configuration, transport mechanisms, and endpoints, simplifying the complexities of interacting with the message broker. It provides an easy-to-use API for publishing messages to the message broker.

The code for the worker class is shown below:

public class Worker : BackgroundService
{
    private readonly ILogger<Worker> _logger;
    private readonly IBus _bus;

    public Worker(ILogger<Worker> logger, IBus bus)
    {
        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
        _bus = bus ?? throw new ArgumentNullException(nameof(bus));
    }

    protected override async Task ExecuteAsync(CancellationToken cancellationToken)
    {
        while (!cancellationToken.IsCancellationRequested)
        {
            if (_logger.IsEnabled(LogLevel.Information))
            {
                _logger.LogInformation("Worker running at: {time}", DateTimeOffset.Now);
            }

            var data = GetTelemetryDataMessage();

            var serializedData = JsonSerializer.Serialize(data);
            _logger.LogDebug("Publishing telemetry data message: {telemetryDataMessage}", serializedData);

            await _bus.Publish(data, cancellationToken);

            await Task.Delay(3000, cancellationToken);
        }
    }

    public TelemetryDataMessage GetTelemetryDataMessage()
    {
        var random = new Random();
        var deviceIdList = new List<string> { "Device-1", "Device-2", "Device-3" };

        var deviceId = deviceIdList[random.Next(0, deviceIdList.Count)];
        var timestamp = DateTime.Now;

        var telemteryDataMessage = new TelemetryDataMessage()
        {
            DeviceId = deviceId,
            Timestamp = timestamp,
            WaterMeasurementData = GetWaterMeasurementData()
        };

        return telemteryDataMessage;
    }

    private static WaterMeasurementData GetWaterMeasurementData()
    {
        var random = new Random();
        var waterMeasurementData = new WaterMeasurementData
        {
            WaterLevel = random.NextDouble() * 100, // 0 to 100 cm
            pHLevel = Math.Round(random.NextDouble() * 14, 2), // 0 to 14 pH
            Temperature = Math.Round(random.NextDouble() * 40, 2), // 0 to 40 °C
            NitrateConcentration = Math.Round(random.NextDouble() * 50, 2), // 0 to 50 mg/L
            DataQuality = "High"
        };

        // Occasionally inject anomalies
        if (random.Next(0, 10) < 3) // 30% chance to inject an anomaly
        {
            waterMeasurementData = InjectAnomaly(waterMeasurementData);
            waterMeasurementData.DataQuality = "Low";
        }

        return waterMeasurementData;
    }

    private static WaterMeasurementData InjectAnomaly(WaterMeasurementData waterMeasurementData)
    {
        var random = new Random();
        var anomalyType = random.Next(0, 4); // 0 to 3

        switch (anomalyType)
        {
            case 0:
                // Inject an anomaly for WaterLevel (e.g., negative or excessively high value)
                waterMeasurementData.WaterLevel = random.NextDouble() > 0.5 ? -random.NextDouble() * 100 : random.NextDouble() * 200;
                break;
            case 1:
                // Inject an anomaly for pHLevel (e.g., negative or above 14)
                waterMeasurementData.pHLevel = random.NextDouble() > 0.5 ? -Math.Round(random.NextDouble() * 14, 2) : Math.Round(random.NextDouble() * 28, 2);
                break;
            case 2:
                // Inject an anomaly for Temperature (e.g., negative or excessively high value)
                waterMeasurementData.Temperature = random.NextDouble() > 0.5 ? -Math.Round(random.NextDouble() * 40, 2) : Math.Round(random.NextDouble() * 80, 2);
                break;
            case 3:
                // Inject an anomaly for NitrateConcentration (e.g., negative or excessively high value)
                waterMeasurementData.NitrateConcentration = random.NextDouble() > 0.5 ? -Math.Round(random.NextDouble() * 50, 2) : Math.Round(random.NextDouble() * 100, 2);
                break;
        }

        return waterMeasurementData;
    }
}

Now that we have our message and publisher (or producer) ready, we will create a consumer class for the message.

When working with MassTransit, it’s natural to wonder whether each published message type has its queue or if all messages are sent to the same queue. The short answer: no, published messages do not have their own dedicated queues. Instead, the routing is handled dynamically based on the publish-subscribe model, ensuring that messages are delivered only to the consumers interested in them.

MassTransit follows the publish-subscribe pattern for message distribution. Here’s a breakdown of the process:

1. Exchanges and Topics: When a message is published, it’s sent to an exchange in RabbitMQ . This exchange is typically named after the message type, and it’s responsible for routing the message to the appropriate consumer queues.
2. Consumer Queues: Each consumer has its own queue. For example:
   • A consumer listening for OrderSubmitted messages will have a queue bound to the OrderSubmitted exchange.
   • Another consumer listening for PaymentProcessed messages will have a different queue bound to the PaymentProcessed exchange.
3. Dynamic Routing: MassTransit uses the message type as a routing key. When you publish a message, only the queues subscribed to that specific message type receive the message. This ensures that messages are delivered only to the relevant consumers.

To consume messages from RabbitMQ (or any supported message broker), you need to create a message consumer. Specifically, a message consumer is a class responsible for handling one or more message types. For each message type, the class implements IConsumer<TMessage> and defines the logic in the Consume method to process the received messages.

As a practical example, the TelemetryService is a standard console application that consumes messages from RabbitMQ. To implement this, we register the message consumer class in the AddMassTransit method during the startup configuration. Consequently, this ensures that the consumer is properly wired to receive and process messages from the queue.

builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<TelemetryDataConsumer>();
    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("localhost", "/", host =>
        {
            host.Username("guest");
            host.Password("guest");
        });

        cfg.ConfigureEndpoints(context);
    });
});

builder.Services.AddLogging();
builder.Services.AddSingleton<ITelemetryDataProcessor, TelemetryDataProcessor>();
builder.Services.AddSingleton<IAnomalyDetector, AnomalyDetector>();

var host = builder.Build();
host.Run();

After the service consumes a message from RabbitMQ, the ITelemetryDataProcessor takes over to validate and process the data. If an anomaly is detected during this processing, the service publishes a new anomaly message back to RabbitMQ. This anomaly message is then picked up and handled by another service specifically designed for anomaly management.

This approach demonstrates how you can build modular, event-driven architectures using RabbitMQ and MassTransit, where each service has a clearly defined responsibility and communicates through messaging.

The consumer class looks like this:

public sealed class TelemetryDataConsumer : IConsumer<TelemetryDataMessage>
{
    private readonly ILogger<TelemetryDataConsumer> _logger;
    private readonly ITelemetryDataProcessor _telemtryDataProcessor;

    public TelemetryDataConsumer(ILogger<TelemetryDataConsumer> logger, ITelemetryDataProcessor telemtryDataProcessor)
    {
        _logger = logger;
        _telemtryDataProcessor = telemtryDataProcessor;
    }

    public async Task Consume(ConsumeContext<TelemetryDataMessage> context)
    {
        if (_logger.IsEnabled(LogLevel.Debug))
        {
            _logger.LogDebug("Consuming telemetry data message: {telemetryDataMessage}", context.Message);
        }

        if (!await ValidateMessageAsync(context))
        {
            return;
        }

        await _telemtryDataProcessor.Process(context.Message);

        if (_logger.IsEnabled(LogLevel.Information))
        {
            _logger.LogInformation("Telemetry data message processed: {telemetryDataMessage}", context.Message);
        }
    }

    private async Task<bool> ValidateMessageAsync(ConsumeContext<TelemetryDataMessage> context)
    {
        if (context.Message == null)
        {
            _logger.LogError("Received null telemetry data message.");
            await context.Publish(new DeadLetterEvent(
                Timestamp: DateTime.UtcNow,
                Reason: "Null message",
                OriginalMessage: null));
            return false;
        }

        if (string.IsNullOrWhiteSpace(context.Message.DeviceId) ||
            context.Message.Timestamp == default ||
            context.Message.WaterMeasurementData == null)
        {
            _logger.LogWarning("Invalid telemetry data for DeviceId: {DeviceId} at {Timestamp}",
                context.Message.DeviceId, context.Message.Timestamp);
            await context.Publish(new DeadLetterEvent(
                Timestamp: DateTime.UtcNow,
                Reason: "Invalid telemetry data",
                OriginalMessage: context.Message));
            return false;
        }

        return true;
    }
}

This section covered the basics of using MassTransit and RabbitMQ to build an event-driven telemetry system. We set up a producer to publish sensor data, a consumer to process it, and logic to handle anomalies. With the foundation laid, we’ll now explore more advanced use cases and techniques.

Here’s how you can set up MassTransit in your .NET application to connect to RabbitMQ securely over SSL:

builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<TelemetryDataConsumer>(configure =>
    {
        configure.UseMessageRetry(r => r.Immediate(5));
    });
    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("localhost", "/", host =>
        {
            host.Username("guest");
            host.Password("guest");

            host.UseSsl(ssl =>
            {
                ssl.ServerName = "localhost";
                ssl.CertificatePassphrase = "password";
                ssl.CertificateValidationCallback = (sender, certificate, chain, sslPolicyErrors) =>
                {
                    // Optional: Validate the certificate. Return true to trust the server certificate.
                    return true;
                };
            });
        });

        cfg.ConfigureEndpoints(context);
    });
});

The host.UseSsl method configures SSL settings, including:

• ServerName: This must match the Common Name (CN) or Subject Alternative Name (SAN) in the server certificate.
• CertificateValidationCallback: Allows custom handling of certificate validation, useful for self-signed certificates.
• AllowPolicyErrors: Relaxes validation policies, which can be useful in development but should be avoided in production.

MassTransit allows us to manage bus startup and shutdown behavior through the MassTransitHostOptions configuration. These options provide control over how the application handles broker connections, timing for startup and shutdown, and long-running consumers

The following options can be adjusted to suit your application’s needs:

Below is an example of how to configure these options using the Options pattern in a .NET application:

builder.Services.AddOptions<MassTransitHostOptions>()
    .Configure<IServiceProvider>((options, sp) =>
    {
        options.WaitUntilStarted = false;
        options.StartTimeout = TimeSpan.FromSeconds(10);
        options.StopTimeout = TimeSpan.FromSeconds(10);
        options.ConsumerStopTimeout = TimeSpan.FromSeconds(10);
    });

MassTransit provides several types of filters to enable middleware-like processing of messages. Filters are components that intercept messages at various stages of the pipeline to perform tasks like logging, validation, transformation, or other custom logic. Let’s review what filters does MassTransit has to offer and how to use them.

The KillSwitch Filter in MassTransit is a special type of filter used to abort message processing under certain conditions. When a “kill switch” is triggered, it stops further message processing and can halt the consumption of messages, effectively preventing any further work from being done by the consumers.

(Reference for the example from the official documentation here)

builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<TelemetryDataConsumer>(configure =>
    {
        configure.UseMessageRetry(r => r.Immediate(5));
    });
    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("localhost", "/", host =>
        {
            host.Username("guest");
            host.Password("guest");
        });

        cfg.ConfigureEndpoints(context);

        // filters
        cfg.UseKillSwitch(options =>
        {
            options.SetActivationThreshold(10);
            options.SetTripThreshold(0.15);
            options.SetRestartTimeout(TimeSpan.FromSeconds(30));
        });
    });
});

n MassTransit, the Circuit Breaker Filter is a built-in safety mechanism designed to prevent your system from overloading when downstream components are struggling. It monitors failures in the message pipeline and, when too many errors occur within a defined time frame, it “trips” the circuit, halting message processing temporarily.

This pause in processing ensures that the failing component is not overwhelmed with retries or additional load, giving it time to recover. Once the reset interval passes, the circuit “closes,” and normal message processing resumes.

Why Use It?

Imagine a scenario where one of your consumers depends on a database that suddenly becomes unresponsive. Without a Circuit Breaker, MassTransit might continue sending messages to the consumer, causing repeated failures and potentially cascading issues throughout your system. The Circuit Breaker Filter steps in to stop this cycle by halting further processing until the database stabilizes.

cfg.UseCircuitBreaker(cb =>
{
    cb.TrackingPeriod = TimeSpan.FromMinutes(1);
    cb.TripThreshold = 20;
    cb.ActiveThreshold = 10;
    cb.ResetInterval = TimeSpan.FromMinutes(5);
});

For more information and details on MassTransit filters, please refer to the official documentation.

• Isolate Message Models: Maintain a separate project or assembly for your message contracts. This ensures consistency between publishers and consumers, facilitating easier updates and versioning.
• Implement Singleton Bus Instances: Initialize the MassTransit bus as a singleton to prevent multiple connections to RabbitMQ, reducing resource consumption and potential connection issues.
• Design Thoughtful Routing Topologies: Plan your exchange and queue structures carefully to ensure efficient message routing. Implement dead-letter queues to handle message failures gracefully, and avoid routing all messages to a single queue to prevent bottlenecks.
• Utilize RabbitMQ’s capabilities such as publish confirmations and durable queues to enhance message reliability and durability.

Working with MassTransit and RabbitMQ provides a powerful combination for building scalable, reliable, and maintainable distributed systems. By understanding patterns like the Circuit Breaker and following best practices for configuration, observability, and resource management, you can ensure your messaging infrastructure is robust and resilient to failure.

Remember, the key to success lies in planning and iterating—designing thoughtful message contracts, leveraging built-in features, and monitoring your system’s health continuously. With these tools and techniques, you’re well-equipped to handle the challenges of modern messaging systems while keeping your applications running smoothly.

Cover Photo by Pixabay