Consumer Configuration

Consumers

To understand consumers and how to create one, refer to the Consumers section.

High-level concepts covered in this configuration section include:

Concept	Description
Consumer	A class that consumes one or more messages types, one for each implementation of IConsumer<TMessage>
Batch Consumer	A class that consumes multiple messages in batches, by implementing IConsumer<Batch<TMessage>>
Job Consumer	A class that consumes a job message, specified by the IJobConsumer<TJob> interface
Consumer Definition	A class, derived from ConsumerDefinition<TConsumer> that configures settings and the consumer’s receive endpoint
Receive Endpoint	Receives messages from a broker queue and delivers those messages to consumer types configured on the receive endpoint

Consumers can be added many ways allowing either a simple of fine-grained approach to registration. Consumers are added inside the AddMassTransit configuration, but before the transport.

using MassTransit;

services.AddMassTransit(x =>
{
    x.AddConsumer<MyConsumer>();

    x.Using[Transport]((context, cfg) =>
    {
        // transport, middleware, other configuration

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

Adding Consumers

Adds a single consumer, with all defaults

AddConsumer<MyConsumer>();
AddConsumer(typeof(MyConsumer));

Adds a consumer, with a consumer definition.

AddConsumer<MyConsumer, MyConsumerDefinition>();
AddConsumer(typeof(MyConsumer), typeof(MyConsumerDefinition));

Adds a consumer with a matching consumer definition and configures the consumer pipeline.

AddConsumer<MyConsumer, MyConsumerDefinition>(cfg =>
{
    cfg.ConcurrentMessageLimit = 8;
});

Adds the specified consumers and consumer definitions. When consumer definitions are included they will be added with the matching consumer type.

void AddConsumers(params Type[] types);

Adds all consumers and consumer definitions in the specified an assembly or assemblies.

void AddConsumers(params Assembly[] assemblies);

Adds the consumers and any matching consumer definitions in the specified an assembly or assemblies that pass the filter. The filter is only called for consumer types.

void AddConsumers(Func<Type, bool> filter, params Assembly[] assemblies);

Configure the endpoint

The consumer endpoint can be configured using the AddConsumer method by calling the Endpoint extension method. For more details, see the endpoint configuration section.

Configure batch consumer options

If you want your consumer to process multiple messages at a time, you can configure a Batch Consumer. This is a consumer that implements IConsumer<Batch<TMessage>>.

AddConsumer<MyBatchConsumer>(cfg =>
{
    cfg.Options<BatchOptions>(options => options
        .SetMessageLimit(100)
        .SetTimeLimit(s: 1)
        .SetTimeLimitStart(BatchTimeLimitStart.FromLast)
        .GroupBy<MyMessage, string>(x => x.CustomerId)
        .SetConcurrencyLimit(10));
});

Property	Type	Default	Description
MessageLimit	int	10	Max number of messages in a batch
ConcurrencyLimit	int	1	number of concurrent batches
TimeLimit	TimeSpan	1 sec	maximum time to wait before delivering a partial batch
TimeLimitStart	TimeSpan	From First	starting point
GroupKeyProvider	object?	null	the property to group by

Configure job options

If you are configuring a job consumer, the job options should be configured. See the job consumers section for details.

Configure Endpoints

By default, MassTransit can automatically configure receive endpoints for all consumers by calling ConfigureEndpoints. You can customize this behavior using a ConsumerDefinition (see below)[#create-a-consumer-definition] or by specifying the endpoint configuration as part of the AddConsumer method as shown above

using MassTransit;
services.AddMassTransit(x =>
{
    // Step 1: Add Consumers Here

    // Step 2: Select a Transport
    x.Using[Transport]((context, cfg) => {
        // Step 3: Configure the Transport

        // Step 4: Configure Endpoints
        // All consumers registered in step 1, will get
        // default endpoints created.
        cfg.ConfigureEndpoints(context);
    });
});

Customized Endpoints

To manually configure a consumer on a receive endpoint, use one of the following methods. You may want to do this for the following reasons.

• Group Consumers onto a specific queue, vs the default of one queue per consumer

Note

Order Matters: Manually configured receive endpoints should be configured before calling ConfigureEndpoints.

cfg.ReceiveEndpoint("manually-configured", e =>
{
    // configure endpoint-specific settings first
    e.SomeEndpointSetting = someValue;

    // configure any required middleware components next
    e.UseMessageRetry(r => r.Interval(5, 1000));

    // configure the consumer last
    e.ConfigureConsumer<MyConsumer>(context);
});

// configure any remaining consumers, sagas, etc.
cfg.ConfigureEndpoints(context);

Endpoint Configuration is Custom by Transport

• RabbitMQ
• Azure Service Bus
• Amazon SQS

Consumer Configuration

ConfigureConsumer<T>(context);

Configures the consumer on the receive endpoint.

ConfigureConsumer<T>(context, consumer =>
{
    // configure consumer-specific middleware
});

Configures the consumer on the receive endpoint and applies the additional consumer configuration to the consumer pipeline.

ConfigureConsumers(context);

Configures all consumers that haven’t been configured on the receive endpoint.

Create a Consumer Definition

Consumer definitions are used to configure consumers and their endpoints. Definitions can be configured explicitly using the AddConsumer method and are also discovered automatically using any of the assembly/namespace scanning AddConsumers methods.

The constructor can be used to override the default endpoint name or to limit the number of messages consumed concurrently. The ConfigureConsumer method is used to configure both the Receive endpoint and the consumer pipeline.

public class SubmitOrderConsumer :
    IConsumer<SubmitOrder>
{
    readonly ILogger<SubmitOrderConsumer> _logger;

    public SubmitOrderConsumer(ILogger<SubmitOrderConsumer> logger)
    {
        _logger = logger;
    }

    public async Task Consume(ConsumeContext<SubmitOrder> context)
    {
        _logger.LogInformation("Order Submitted: {OrderId}", context.Message.OrderId);

        await context.Publish<OrderSubmitted>(new
        {
            context.Message.OrderId
        });
    }
}

public class SubmitOrderConsumerDefinition :
    ConsumerDefinition<SubmitOrderConsumer>
{
    public SubmitOrderConsumerDefinition()
    {
        // override the default endpoint name
        EndpointName = "order-service";

        // limit the number of messages consumed concurrently
        // this applies to the consumer only, not the endpoint
        ConcurrentMessageLimit = 4;
    }

    protected override void ConfigureConsumer(IReceiveEndpointConfigurator endpointConfigurator,
        IConsumerConfigurator<SubmitOrderConsumer> consumerConfigurator, IRegistrationContext context)
    {
        // configure message retry with millisecond intervals
        endpointConfigurator.UseMessageRetry(r => r.Intervals(100,200,500,800,1000));

        // use the outbox to prevent duplicate events from being published
        endpointConfigurator.UseInMemoryOutbox(context);
    }
}

In this example, the consumer will be configured to use message retry, an in-memory outbox, and will be limited to four messages consumed concurrently.

Configure endpoint options

Concept	Description
EndpointDefinition	Used internally to override an endpoint definition (commonly used for configuring multiple consumers on a shared endpoint)
EndpointName	The name of the queue for this consumer
ConcurrentMessageLimit	The number of messages this consumer can process concurrently