SemianiakaVY/elsa-core
1
0
Fork 0
Code Issues Pull requests Actions 1 Packages Projects Releases Wiki Activity

Add settings for MassTransit message handling (#5176)

Browse source 
* Added settings for increasing message handling throughput

* Fixed automatic deletion of queues and exchanges in rabbitMQ
This commit is contained in:
raymonddenhaan 2024-04-03 23:32:50 +02:00 committed by GitHub
parent d146c5d938
commit afb755cf52
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 52 additions and 31 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

34
src/bundles/Elsa.ServerAndStudio.Web/Program.cs
View file

@ -24,10 +24,12 @@ var azureServiceBusConnectionString = configuration.GetConnectionString("AzureSe
var identitySection = configuration.GetSection("Identity");
var identityTokenSection = identitySection.GetSection("Tokens");
var massTransitSection = configuration.GetSection("MassTransit");
var massTransitDispatcherSection = configuration.GetSection("MassTransit.Dispatcher");
var heartbeatSection = configuration.GetSection("Heartbeat");
const MassTransitBroker useMassTransitBroker = MassTransitBroker.Memory;
const MassTransitBroker useMassTransitBroker = MassTransitBroker.AzureServiceBus;
services.Configure<MassTransitWorkflowDispatcherOptions>(massTransitSection);
services.Configure<MassTransitOptions>(massTransitSection);
services.Configure<MassTransitWorkflowDispatcherOptions>(massTransitDispatcherSection);
// Add Elsa services.
services
@ -87,28 +89,14 @@ services
{
elsa.UseMassTransit(massTransit =>
{
if (useMassTransitBroker == MassTransitBroker.AzureServiceBus)
switch (useMassTransitBroker)
{
massTransit.UseAzureServiceBus(azureServiceBusConnectionString, serviceBusFeature => serviceBusFeature.ConfigureServiceBus = bus =>
{
bus.PrefetchCount = 4;
bus.LockDuration = TimeSpan.FromMinutes(5);
bus.MaxConcurrentCalls = 32;
bus.MaxDeliveryCount = 8;
// etc.
});
}
if (useMassTransitBroker == MassTransitBroker.RabbitMq)
{
massTransit.UseRabbitMq(rabbitMqConnectionString, rabbit => rabbit.ConfigureServiceBus = bus =>
{
bus.PrefetchCount = 4;
bus.Durable = true;
bus.AutoDelete = false;
bus.ConcurrentMessageLimit = 32;
// etc.
});
case MassTransitBroker.AzureServiceBus:
massTransit.UseAzureServiceBus(azureServiceBusConnectionString);
break;
case MassTransitBroker.RabbitMq:
massTransit.UseRabbitMq(rabbitMqConnectionString);
break;
}
}
);

8
src/bundles/Elsa.ServerAndStudio.Web/appsettings.json
View file

@ -66,7 +66,13 @@
"Timeout": "00:00:05:00"
},
"MassTransit": {
"TemporaryQueueTtl": "00:00:05:00"
"TemporaryQueueTtl": "00:00:05:00",
"ConcurrentMessageLimit": 4,
"PrefetchCount": 4,
"MaxAutoRenewDuration": "00:00:00:30",
"Dispatcher": {
"ConcurrentMessageLimit": 4
}
},
"Smtp": {
"Host": "localhost",

9
src/modules/Elsa.MassTransit.AzureServiceBus/Features/AzureServiceBusFeature.cs
View file

@ -73,10 +73,17 @@ public class AzureServiceBusFeature : FeatureBase
if (ConnectionString != null)
configurator.Host(ConnectionString);
var options = context.GetRequiredService<IOptions<MassTransitOptions>>().Value;
if (options.PrefetchCount is not null)
configurator.PrefetchCount = options.PrefetchCount.Value;
if (options.MaxAutoRenewDuration is not null)
configurator.MaxAutoRenewDuration = options.MaxAutoRenewDuration.Value;
configurator.ConcurrentMessageLimit = options.ConcurrentMessageLimit;
configurator.UseServiceBusMessageScheduler();
configurator.SetupWorkflowDispatcherEndpoints(context);
ConfigureServiceBus?.Invoke(configurator);
var options = context.GetRequiredService<IOptions<MassTransitWorkflowDispatcherOptions>>().Value;
var instanceNameProvider = context.GetRequiredService<IApplicationInstanceNameProvider>();
foreach (var consumer in temporaryConsumers)

11
src/modules/Elsa.MassTransit.RabbitMq/Features/RabbitMqServiceBusFeature.cs
View file

@ -56,24 +56,27 @@ public class RabbitMqServiceBusFeature : FeatureBase
configure.UsingRabbitMq((context, configurator) =>
{
var options = context.GetRequiredService<IOptions<MassTransitWorkflowDispatcherOptions>>().Value;
var options = context.GetRequiredService<IOptions<MassTransitOptions>>().Value;
var instanceNameProvider = context.GetRequiredService<IApplicationInstanceNameProvider>();
if (!string.IsNullOrEmpty(ConnectionString))
configurator.Host(ConnectionString);
if (options.PrefetchCount is not null)
configurator.PrefetchCount = options.PrefetchCount.Value;
configurator.ConcurrentMessageLimit = options.ConcurrentMessageLimit;
ConfigureServiceBus?.Invoke(configurator);
foreach (var consumer in temporaryConsumers)
{
configure.AddConsumer(consumer.ConsumerType).ExcludeFromConfigureEndpoints();
configurator.ReceiveEndpoint($"{instanceNameProvider.GetName()}-{consumer.Name}",
endpointConfigurator =>
{
endpointConfigurator.QueueExpiration = options.TemporaryQueueTtl ?? TimeSpan.FromHours(1);
endpointConfigurator.ConcurrentMessageLimit = options.ConcurrentMessageLimit;
endpointConfigurator.Durable = false;
endpointConfigurator.AutoDelete = true;
endpointConfigurator.ConfigureConsumer(context, consumer.ConsumerType);
});
}

2
src/modules/Elsa.MassTransit/Features/MassTransitFeature.cs
View file

@ -33,6 +33,7 @@ public class MassTransitFeature : FeatureBase
}
/// The number of messages to prefetch.
[Obsolete("PrefetchCount has been moved to be included in MassTransitOptions")]
public int? PrefetchCount { get; set; }
/// <summary>
@ -56,6 +57,7 @@ public class MassTransitFeature : FeatureBase
var messageTypes = this.GetMessages();
Services.AddSingleton(ChannelQueueFormatterFactory);
Services.Configure<MassTransitOptions>(x => x.PrefetchCount ??= PrefetchCount);
Services.Configure<MassTransitWorkflowDispatcherOptions>(x => { });
Services.AddActivityProvider<MassTransitActivityTypeProvider>();
_runInMemory = BusConfigurator is null;

17
src/modules/Elsa.MassTransit/Options/MassTransitOptions.cs Normal file
View file

@ -0,0 +1,17 @@
namespace Elsa.MassTransit.Options;
/// <summary>
/// Represents the options for configuring MassTransit.
/// </summary>
public class MassTransitOptions
{
/// The TTL of queues that are seen as temporary (typically queues that are created per running instance).
public TimeSpan? TemporaryQueueTtl { get; set; }
/// The number of concurrent messages to process.
public int? ConcurrentMessageLimit { get; set; }
/// The number of messages to fetch from the bus with each request.
public int? PrefetchCount { get; set; }
/// The maximum duration for auto-renewal of a resource.
/// <remarks>Only relevant when using Azure Service Bus.</remarks>
public TimeSpan? MaxAutoRenewDuration { get; set; }
}

2
src/modules/Elsa.MassTransit/Options/MassTransitWorkflowDispatcherOptions.cs
View file

@ -5,8 +5,6 @@ namespace Elsa.MassTransit.Options;
/// Provides options to the <see cref="DispatchWorkflowRequestConsumerDefinition"/>
public class MassTransitWorkflowDispatcherOptions
{
/// The TTL of queues that are seen as temporary (typically queues that are created per running instance).
public TimeSpan? TemporaryQueueTtl { get; set; }
/// The number of concurrent messages to process.
public int? ConcurrentMessageLimit { get; set; }
}