Add scheduler batch size option (#1689) (#1704)

Author: yang-xiaodong

docs/content/user-guide/en/cap/configuration.md

@@ -146,7 +146,13 @@ If set to true, we will use a database-based distributed lock to solve the probl
 
 The interval of the collector processor deletes expired messages.
 
-#### ConsumerThreadCount 
+#### SchedulerBatchSize
+
+> Default: 1000
+
+Maximum number of delayed or queued messages fetched per scheduler cycle.
+
+#### ConsumerThreadCount
 
 > Default: 1
 

docs/content/user-guide/zh/cap/configuration.md

@@ -152,6 +152,12 @@ Group 在不同的 Broker 有不同的对应项。
 
 收集器删除已经过期消息的时间间隔。
 
+#### SchedulerBatchSize
+
+> 默认值：1000
+
+调度器每次循环获取的延迟或排队消息的最大数量。
+
 #### FailedRetryCount
 
 > 默认值：50

src/DotNetCore.CAP.InMemoryStorage/IDataStorage.InMemory.cs

@@ -228,6 +228,7 @@ public Task ScheduleMessagesOfDelayedAsync(Func<object, IEnumerable<MediumMessag
         var result = PublishedMessages.Values.Where(x =>
                 (x.StatusName == StatusName.Delayed && x.ExpiresAt < DateTime.Now.AddMinutes(2))
                 || (x.StatusName == StatusName.Queued && x.ExpiresAt < DateTime.Now.AddMinutes(-1)))
+            .Take(_capOptions.Value.SchedulerBatchSize)
             .Select(x => (MediumMessage)x);
 
         return scheduleTask(null!, result);

src/DotNetCore.CAP.MongoDB/IDataStorage.MongoDB.cs

@@ -336,7 +336,9 @@ public async Task ScheduleMessagesOfDelayedAsync(Func<object, IEnumerable<Medium
                     await collection.UpdateManyAsync(session, filter, update, cancellationToken: linkedTs.Token)
                         .ConfigureAwait(false);
 
-                    var queryResult = await collection.Find(session, filter).ToListAsync(linkedTs.Token)
+                    var queryResult = await collection.Find(session, filter)
+                        .Limit(_capOptions.Value.SchedulerBatchSize)
+                        .ToListAsync(linkedTs.Token)
                         .ConfigureAwait(false);
 
                     var result = queryResult.Select(x => new MediumMessage

src/DotNetCore.CAP.MySql/IDataStorage.MySql.cs

@@ -263,13 +263,14 @@ public async Task ScheduleMessagesOfDelayedAsync(Func<object, IEnumerable<Medium
 
         var sql =
             $"SELECT `Id`,`Content`,`Retries`,`Added`,`ExpiresAt` FROM `{_pubName}` WHERE `Version`=@Version " +
-            $"AND ((`ExpiresAt`< @TwoMinutesLater AND `StatusName` = '{StatusName.Delayed}') OR (`ExpiresAt`< @OneMinutesAgo AND `StatusName` = '{StatusName.Queued}')) {lockSql};";
+            $"AND ((`ExpiresAt`< @TwoMinutesLater AND `StatusName` = '{StatusName.Delayed}') OR (`ExpiresAt`< @OneMinutesAgo AND `StatusName` = '{StatusName.Queued}')) LIMIT @BatchSize {lockSql};";
 
         object[] sqlParams =
         {
             new MySqlParameter("@Version", _capOptions.Value.Version),
             new MySqlParameter("@TwoMinutesLater", DateTime.Now.AddMinutes(2)),
-            new MySqlParameter("@OneMinutesAgo", DateTime.Now.AddMinutes(-1))
+            new MySqlParameter("@OneMinutesAgo", DateTime.Now.AddMinutes(-1)),
+            new MySqlParameter("@BatchSize", _capOptions.Value.SchedulerBatchSize)
         };
 
         await using var connection = new MySqlConnection(_options.Value.ConnectionString);

src/DotNetCore.CAP.PostgreSql/IDataStorage.PostgreSql.cs

@@ -257,13 +257,14 @@ public async Task ScheduleMessagesOfDelayedAsync(Func<object, IEnumerable<Medium
     {
         var sql =
             $"SELECT \"Id\",\"Content\",\"Retries\",\"Added\",\"ExpiresAt\" FROM {_pubName} WHERE \"Version\"=@Version " +
-            $"AND ((\"ExpiresAt\"< @TwoMinutesLater AND \"StatusName\" = '{StatusName.Delayed}') OR (\"ExpiresAt\"< @OneMinutesAgo AND \"StatusName\" = '{StatusName.Queued}')) FOR UPDATE SKIP LOCKED;";
+            $"AND ((\"ExpiresAt\"< @TwoMinutesLater AND \"StatusName\" = '{StatusName.Delayed}') OR (\"ExpiresAt\"< @OneMinutesAgo AND \"StatusName\" = '{StatusName.Queued}')) FOR UPDATE SKIP LOCKED LIMIT @BatchSize;";
 
         var sqlParams = new object[]
         {
             new NpgsqlParameter("@Version", _capOptions.Value.Version),
             new NpgsqlParameter("@TwoMinutesLater", DateTime.Now.AddMinutes(2)),
-            new NpgsqlParameter("@OneMinutesAgo", QueuedMessageFetchTime())
+            new NpgsqlParameter("@OneMinutesAgo", QueuedMessageFetchTime()),
+            new NpgsqlParameter("@BatchSize", _capOptions.Value.SchedulerBatchSize)
         };
 
         await using var connection = _options.Value.CreateConnection();

src/DotNetCore.CAP.SqlServer/IDataStorage.SqlServer.cs

@@ -255,17 +255,18 @@ public async Task ScheduleMessagesOfDelayedAsync(Func<object, IEnumerable<Medium
         CancellationToken token = default)
     {
         var sql = $@"
-            SELECT Id, Content, Retries, Added, ExpiresAt FROM {_pubName} WITH (UPDLOCK, READPAST)
+            SELECT TOP (@BatchSize) Id, Content, Retries, Added, ExpiresAt FROM {_pubName} WITH (UPDLOCK, READPAST)
                 WHERE Version = @Version AND StatusName = '{StatusName.Delayed}' AND ExpiresAt < @TwoMinutesLater
             UNION ALL
-            SELECT Id, Content, Retries, Added, ExpiresAt FROM {_pubName} WITH (UPDLOCK, READPAST)
+            SELECT TOP (@BatchSize) Id, Content, Retries, Added, ExpiresAt FROM {_pubName} WITH (UPDLOCK, READPAST)
                 WHERE Version = @Version AND StatusName = '{StatusName.Queued}' AND ExpiresAt < @OneMinutesAgo;";
 
         object[] sqlParams =
         {
             new SqlParameter("@Version", _capOptions.Value.Version),
             new SqlParameter("@TwoMinutesLater", DateTime.Now.AddMinutes(2)),
-            new SqlParameter("@OneMinutesAgo", DateTime.Now.AddMinutes(-1))
+            new SqlParameter("@OneMinutesAgo", DateTime.Now.AddMinutes(-1)),
+            new SqlParameter("@BatchSize", _capOptions.Value.SchedulerBatchSize)
         };
 
         await using var connection = new SqlConnection(_options.Value.ConnectionString);

src/DotNetCore.CAP/CAP.Options.cs

@@ -35,6 +35,7 @@ public CapOptions()
         DefaultGroupName = "cap.queue." + Assembly.GetEntryAssembly()?.GetName().Name!.ToLower();
         CollectorCleaningInterval = 300;
         FallbackWindowLookbackSeconds = 240;
+        SchedulerBatchSize = 1000;
     }
 
     internal IList<ICapOptionsExtension> Extensions { get; }
@@ -140,6 +141,12 @@ public CapOptions()
     /// </summary>
     public int CollectorCleaningInterval { get; set; }
 
+    /// <summary>
+    /// Maximum number of delayed or queued messages fetched per scheduler cycle.
+    /// Default is 1000.
+    /// </summary>
+    public int SchedulerBatchSize { get; set; }
+
     /// <summary>
     /// Configure JSON serialization settings
     /// </summary>