Autotest
/
sharding
mirror of https://gitee.com/dotnetchina/sharding-core.git
1
0
Fork 0
Code Issues Releases Wiki Activity

优化并发查询控制使用异步

This commit is contained in:
xuejiaming 2021-11-30 09:53:07 +08:00
parent a317ceed0d
commit d91dd52c5e
3 changed files with 14 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
samples/Sample.SqlServer/Startup.cs
View File

@ -7,13 +7,9 @@ using Microsoft.Extensions.Logging;
using Sample.SqlServer.DbContexts;
using Sample.SqlServer.Shardings;
using ShardingCore;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using Sample.SqlServer.Domain.Entities;
using ShardingCore.Extensions;
using ShardingCore.Sharding.ReadWriteConfigurations;
using ShardingCore.Sharding.ShardingComparision;
using System;
using System.Collections.Generic;
namespace Sample.SqlServer
{

22
src/ShardingCore/Sharding/MergeEngines/Abstractions/AbstractBaseMergeEngine.cs
View File

@ -38,6 +38,7 @@ namespace ShardingCore.Sharding.MergeEngines.Abstractions
_semaphore = new SemaphoreSlim(Math.Max(1, streamMergeContext.GetParallelQueryMaxThreadCount()));
_parallelQueryTimeOut = streamMergeContext.GetParallelQueryTimeOut();
}
/// <summary>
/// 异步多线程控制并发
/// </summary>
@ -45,17 +46,14 @@ namespace ShardingCore.Sharding.MergeEngines.Abstractions
/// <param name="executeAsync"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
public Task<TResult> AsyncParallelLimitExecuteAsync<TResult>(Func<Task<TResult>> executeAsync,CancellationToken cancellationToken=new CancellationToken())
public async Task<TResult> AsyncParallelLimitExecuteAsync<TResult>(Func<Task<TResult>> executeAsync,CancellationToken cancellationToken=new CancellationToken())
{
cancellationToken.ThrowIfCancellationRequested();
var parallelTimeOut = _parallelQueryTimeOut.TotalMilliseconds;
var acquired = this._semaphore.Wait((int)parallelTimeOut);
var acquired = await this._semaphore.WaitAsync((int)parallelTimeOut, cancellationToken);
if (acquired)
{
var once = new SemaphoreReleaseOnlyOnce(this._semaphore);
try
{
return Task.Run(async () =>
{
try
{
return await executeAsync();
@ -64,18 +62,16 @@ namespace ShardingCore.Sharding.MergeEngines.Abstractions
{
once.Release();
}
}, cancellationToken);
}
catch (Exception e)
{
once.Release();
throw e;
}
}
else
{
throw new ShardingCoreParallelQueryTimeOutException(_executeExpression.ShardingPrint());
}
}
public void Dispose()
{
_semaphore?.Dispose();
}
}
}

2
src/ShardingCore/Sharding/MergeEngines/Abstractions/IAsyncParallelLimit.cs
View File

@ -13,7 +13,7 @@ namespace ShardingCore.Sharding.MergeEngines.Abstractions
* @Ver: 1.0
* @Email: 326308290@qq.com
*/
internal interface IAsyncParallelLimit
internal interface IAsyncParallelLimit:IDisposable
{
Task<TResult> AsyncParallelLimitExecuteAsync<TResult>(Func<Task<TResult>> executeAsync,
CancellationToken cancellationToken = new CancellationToken());