From f4d544c40b1c790d76b896de7bab36f635637888 Mon Sep 17 00:00:00 2001 From: liangshiwei Date: Mon, 25 Jul 2022 16:56:48 +0800 Subject: [PATCH] Enhance RabbitMQ to set PrefetchCount --- docs/en/Background-Jobs-RabbitMq.md | 12 +++++++++--- .../en/Distributed-Event-Bus-RabbitMQ-Integration.md | 3 ++- docs/zh-Hans/Background-Jobs-RabbitMq.md | 10 ++++++++-- .../Distributed-Event-Bus-RabbitMQ-Integration.md | 3 ++- .../RabbitMQ/AbpRabbitMqBackgroundJobOptions.cs | 2 ++ .../Volo/Abp/BackgroundJobs/RabbitMQ/JobQueue.cs | 8 +++++++- .../BackgroundJobs/RabbitMQ/JobQueueConfiguration.cs | 6 ++++-- .../EventBus/RabbitMq/AbpRabbitMqEventBusOptions.cs | 2 ++ .../EventBus/RabbitMq/RabbitMqDistributedEventBus.cs | 3 ++- .../Volo/Abp/RabbitMQ/QueueDeclareConfiguration.cs | 6 +++++- .../Volo/Abp/RabbitMQ/RabbitMqMessageConsumer.cs | 5 +++++ 11 files changed, 48 insertions(+), 12 deletions(-) diff --git a/docs/en/Background-Jobs-RabbitMq.md b/docs/en/Background-Jobs-RabbitMq.md index 16b8b0981a..8eec3733fb 100644 --- a/docs/en/Background-Jobs-RabbitMq.md +++ b/docs/en/Background-Jobs-RabbitMq.md @@ -126,25 +126,31 @@ By default, all the job types use the `Default` RabbitMQ connection. Configure(options => { options.DefaultQueueNamePrefix = "my_app_jobs."; + options.DefaultDelayedQueueNamePrefix = "my_app_jobs.delayed" + options.PrefetchCount = 1; options.JobQueues[typeof(EmailSendingArgs)] = new JobQueueConfiguration( typeof(EmailSendingArgs), queueName: "my_app_jobs.emails", - connectionName: "SecondConnection" + connectionName: "SecondConnection", + delayedQueueName:"my_app_jobs.emails.delayed" ); }); ```` -* This example sets the default queue name prefix to `my_app_jobs.`. If different applications use the same RabbitMQ server, it would be important to use different prefixes for each application to not consume jobs of each other. +* This example sets the default queue name prefix to `my_app_jobs.` and default delayed queue name prefix to `my_app_jobs.delayed`. If different applications use the same RabbitMQ server, it would be important to use different prefixes for each application to not consume jobs of each other. +* Sets `PrefetchCount` for all queues. * Also specifies a different connection string for the `EmailSendingArgs`. `JobQueueConfiguration` class has some additional options in its constructor; * `queueName`: The queue name that is used for this job. The prefix is not added, so you need to specify the full name of the queue. +* `DelayedQueueName`: The delayed queue name that is used for delayed execution of job. The prefix is not added, so you need to specify the full name of the queue. * `connectionName`: The RabbitMQ connection name (see the connection configuration above). This is optional and the default value is `Default`. * `durable` (optional, default: `true`). * `exclusive` (optional, default: `false`). -* `autoDelete` (optional, default: `false`) +* `autoDelete` (optional, default: `false`). +* `PrefetchCount` (optional, default: null) See the RabbitMQ documentation if you want to understand the `durable`, `exclusive` and `autoDelete` options better, while most of the times the default configuration is what you want. diff --git a/docs/en/Distributed-Event-Bus-RabbitMQ-Integration.md b/docs/en/Distributed-Event-Bus-RabbitMQ-Integration.md index 3d99e1e3cf..1961726b8f 100644 --- a/docs/en/Distributed-Event-Bus-RabbitMQ-Integration.md +++ b/docs/en/Distributed-Event-Bus-RabbitMQ-Integration.md @@ -141,13 +141,14 @@ Configure(options => }); ```` -**Example: Configure the client and exchange names** +**Example: Configure the client, exchange names and prefetchCount** ````csharp Configure(options => { options.ClientName = "TestApp1"; options.ExchangeName = "TestMessages"; + options.PrefetchCount = 1; }); ```` diff --git a/docs/zh-Hans/Background-Jobs-RabbitMq.md b/docs/zh-Hans/Background-Jobs-RabbitMq.md index 0f9fb21d8a..8bf61846e7 100644 --- a/docs/zh-Hans/Background-Jobs-RabbitMq.md +++ b/docs/zh-Hans/Background-Jobs-RabbitMq.md @@ -126,25 +126,31 @@ Configure(options => Configure(options => { options.DefaultQueueNamePrefix = "my_app_jobs."; + options.DefaultDelayedQueueNamePrefix = "my_app_jobs.delayed" + options.PrefetchCount = 1; options.JobQueues[typeof(EmailSendingArgs)] = new JobQueueConfiguration( typeof(EmailSendingArgs), queueName: "my_app_jobs.emails", - connectionName: "SecondConnection" + connectionName: "SecondConnection", + delayedQueueName:"my_app_jobs.emails.delayed" ); }); ``` -- 这个示例将默认的队列名前缀设置为 `my_app_jobs.`,如果多个项目都使用的同一个 RabbitMQ 服务,设置不同的前缀可以避免执行其他项目的后台作业. +- 这个示例将默认的队列名前缀设置为 `my_app_jobs.`并且设置默认的延迟队列名为 `my_app_jobs.delayed`,如果多个项目都使用的同一个 RabbitMQ 服务,设置不同的前缀可以避免执行其他项目的后台作业. +- 设置了预取数量, 用于所有队列. - 这里还设置了 `EmailSendingArgs` 绑定的 RabbitMQ 连接. `JobQueueConfiguration` 类的构造函数中,还有一些其他的可选参数. - `queueName`: 指定后台作业对应的队列名称(全名). +* `DelayedQueueName`: 指定后台延迟执行的作业对于的队列名称(全名). - `connectionName`: 后台作业对应的 RabbitMQ 连接名称,默认是 `Default`. - `durable`: 可选参数,默认为 `true`. - `exclusive`: 可选参数,默认为 `false`. - `autoDelete`: 可选参数,默认为 `false`. +* `PrefetchCount` (可选参数, 默认为: null) 如果你想要更多地了解 `durable`,`exclusive`,`autoDelete` 的用法,请阅读 RabbitMQ 提供的文档. diff --git a/docs/zh-Hans/Distributed-Event-Bus-RabbitMQ-Integration.md b/docs/zh-Hans/Distributed-Event-Bus-RabbitMQ-Integration.md index 1cd6b5b9dd..f45d4d90bf 100644 --- a/docs/zh-Hans/Distributed-Event-Bus-RabbitMQ-Integration.md +++ b/docs/zh-Hans/Distributed-Event-Bus-RabbitMQ-Integration.md @@ -141,13 +141,14 @@ Configure(options => }); ```` -**示例: 配置客户端和交换机名称** +**示例: 配置客户端,交换机名称和预取数量** ````csharp Configure(options => { options.ClientName = "TestApp1"; options.ExchangeName = "TestMessages"; + options.PrefetchCount = 1; }); ```` diff --git a/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/AbpRabbitMqBackgroundJobOptions.cs b/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/AbpRabbitMqBackgroundJobOptions.cs index 83efe8ff31..540d848412 100644 --- a/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/AbpRabbitMqBackgroundJobOptions.cs +++ b/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/AbpRabbitMqBackgroundJobOptions.cs @@ -19,6 +19,8 @@ public class AbpRabbitMqBackgroundJobOptions /// Default value: "AbpBackgroundJobsDelayed." /// public string DefaultDelayedQueueNamePrefix { get; set; } + + public ushort? PrefetchCount { get; set; } public AbpRabbitMqBackgroundJobOptions() { diff --git a/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/JobQueue.cs b/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/JobQueue.cs index 208d7ecd9f..59481840aa 100644 --- a/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/JobQueue.cs +++ b/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/JobQueue.cs @@ -66,7 +66,8 @@ public class JobQueue : IJobQueue new JobQueueConfiguration( typeof(TArgs), AbpRabbitMqBackgroundJobOptions.DefaultQueueNamePrefix + JobConfiguration.JobName, - AbpRabbitMqBackgroundJobOptions.DefaultDelayedQueueNamePrefix + JobConfiguration.JobName + AbpRabbitMqBackgroundJobOptions.DefaultDelayedQueueNamePrefix + JobConfiguration.JobName, + prefetchCount: AbpRabbitMqBackgroundJobOptions.PrefetchCount ); } @@ -143,6 +144,11 @@ public class JobQueue : IJobQueue Consumer = new AsyncEventingBasicConsumer(ChannelAccessor.Channel); Consumer.Received += MessageReceived; + if (QueueConfiguration.PrefetchCount.HasValue) + { + ChannelAccessor.Channel.BasicQos(0, QueueConfiguration.PrefetchCount.Value, false); + } + //TODO: What BasicConsume returns? ChannelAccessor.Channel.BasicConsume( queue: QueueConfiguration.QueueName, diff --git a/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/JobQueueConfiguration.cs b/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/JobQueueConfiguration.cs index 792eb9815e..ee706ab864 100644 --- a/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/JobQueueConfiguration.cs +++ b/framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/JobQueueConfiguration.cs @@ -20,12 +20,14 @@ public class JobQueueConfiguration : QueueDeclareConfiguration string connectionName = null, bool durable = true, bool exclusive = false, - bool autoDelete = false) + bool autoDelete = false, + ushort? prefetchCount = null) : base( queueName, durable, exclusive, - autoDelete) + autoDelete, + prefetchCount) { JobArgsType = jobArgsType; ConnectionName = connectionName; diff --git a/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/AbpRabbitMqEventBusOptions.cs b/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/AbpRabbitMqEventBusOptions.cs index addbc49171..bac9bdf0c9 100644 --- a/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/AbpRabbitMqEventBusOptions.cs +++ b/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/AbpRabbitMqEventBusOptions.cs @@ -13,6 +13,8 @@ public class AbpRabbitMqEventBusOptions public string ExchangeName { get; set; } public string ExchangeType { get; set; } + + public ushort? PrefetchCount { get; set; } public string GetExchangeTypeOrDefault() { diff --git a/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs index c9ec49bc7e..7e352ca530 100644 --- a/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs @@ -78,7 +78,8 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe AbpRabbitMqEventBusOptions.ClientName, durable: true, exclusive: false, - autoDelete: false + autoDelete: false, + prefetchCount: AbpRabbitMqEventBusOptions.PrefetchCount ), AbpRabbitMqEventBusOptions.ConnectionName ); diff --git a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/QueueDeclareConfiguration.cs b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/QueueDeclareConfiguration.cs index 36f638668c..4b03c788c6 100644 --- a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/QueueDeclareConfiguration.cs +++ b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/QueueDeclareConfiguration.cs @@ -13,6 +13,8 @@ public class QueueDeclareConfiguration public bool Exclusive { get; set; } public bool AutoDelete { get; set; } + + public ushort? PrefetchCount { get; set; } public IDictionary Arguments { get; } @@ -20,13 +22,15 @@ public class QueueDeclareConfiguration [NotNull] string queueName, bool durable = true, bool exclusive = false, - bool autoDelete = false) + bool autoDelete = false, + ushort? prefetchCount = null) { QueueName = queueName; Durable = durable; Exclusive = exclusive; AutoDelete = autoDelete; Arguments = new Dictionary(); + PrefetchCount = prefetchCount; } public virtual QueueDeclareOk Declare(IModel channel) diff --git a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/RabbitMqMessageConsumer.cs b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/RabbitMqMessageConsumer.cs index 8ce184295a..c6a61b5da2 100644 --- a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/RabbitMqMessageConsumer.cs +++ b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/RabbitMqMessageConsumer.cs @@ -168,6 +168,11 @@ public class RabbitMqMessageConsumer : IRabbitMqMessageConsumer, ITransientDepen var consumer = new AsyncEventingBasicConsumer(Channel); consumer.Received += HandleIncomingMessageAsync; + if (Queue.PrefetchCount.HasValue) + { + Channel.BasicQos(0, Queue.PrefetchCount.Value, false); + } + Channel.BasicConsume( queue: Queue.QueueName, autoAck: false,