使用异步时未释放SqlConnection

Question

我有一个具有Sql-Server数据库后端和Dapper作为ORM的项目。我正在尝试使用Dapper的QueryAsync()方法来获取一些数据。不仅如此，对我的仓库的调用还来自多个用a调用的任务内部Task.WhenAll（也就是说，每个任务都涉及从该仓库中获取数据，因此每个任务都在等待包装该QueryAsync()调用的我仓库的方法）。

问题是即使使用using块，我的SqlConnections也永远不会关闭。结果，我与数据库有100多个开放的连接，最终开始出现“达到最大池大小”异常。问题是，当我切换到just Query()而不是时QueryAsync()，它可以正常工作，但我希望能够异步执行此操作。

这是一个代码示例。我试图尽可能地模仿实际应用的结构，这就是为什么它看起来比必须的更为复杂的原因。

接口：

public interface IFooRepository<T> where T: FooBase
{
    Task<IEnumerable<T>> Select(string account, DateTime? effectiveDate = null);
}

实现方式：

public class FooRepository : RepositoryBase, IFooRepository<SpecialFoo>
{
    private readonly IWebApiClientRepository _accountRepository;

    public FooRepository(IWebApiClientRepository repo)
    {
        _accountRepository = repo;
    }
    public async Task<IEnumerable<FuturePosition>> Select(string code, DateTime? effectiveDate = null)
    {
        effectiveDate = effectiveDate ?? DateTime.Today.Date;
        var referenceData =  await _accountRepository.GetCrossRefferenceData(code, effectiveDate.Value);
        using (var connection = new SqlConnection("iamaconnectionstring")
        {
            connection.Open();
            try
            {
                var res = await connection.QueryAsync<FuturePosition>(SqlQueryVariable + "AND t.code = @code;",
                    new
                    {
                        effectiveDate = effectiveDate.Value,
                        code = referenceData.Code
                    });

                foreach (var item in res)
                {
                    item.PropFromReference = referenceData.PropFromReference;
                }
                return res;
            }
            catch (Exception e)
            {
                //log 
                throw;
            }
            finally
            {
                connection.Close();
            }
        }
    }
}

因此，现在有了调用代码，现在有2层。我将从外部开始。我认为这就是问题所在。在下面有评论。

填充器：

public class Populator : PopulatorBase
{
    private IAccountRepository _acctRepository;
    public override async Task<IEnumerable<PopulationResult>> ProcessAccount(DateTime? popDate = null)
    {
        //My attempt at throttling the async calls
        //I was hoping this would force a max of 10 simultaneous connections.
        //It did not work.
        SemaphoreSlim ss = new SemaphoreSlim(10,10);
        var accountsToProcess = _acctRepository.GetAllAccountsToProcess();
        var accountNumbers = accountsToProcess.SelectMany(a => a.accountNumbers).ToList();

        List<Task<ProcessResult>> trackedTasks = new List<Task<ProcessResult>>();
        foreach (var item in accountNumbers)
        {
            await ss.WaitAsync();
            trackedTasks.Add(ProcessAccount(item.AccountCode, popDate ?? DateTime.Today));
            ss.Release();
        }
        //my gut tells me the issue is because of these tasks
        var results = await Task.WhenAll(trackedTasks);
        return results;
    }

    private async Task<ProcessResult>ProcessAccount(string accountCode, DateTime? popDate)
    {
        var createdItems = await _itemCreator.MakeExceptions(popDate, accountCode);
        return Populate(accountCode, createdItems);
    }
}

ItemCreator：

public class ItemCreator : ItemCreatorBase
{
    private readonly IFooRepository<FuturePosition> _fooRepository;
    private readonly IBarRepository<FuturePosition> _barRepository;

    public RussellGlobeOpFutureExceptionCreator() )
    {
        //standard constructor stuff
    }
    public async Task<ItemCreationResult> MakeItems(DateTime? effectiveDate, string account)
    {
        DateTime reconDate = effectiveDate ?? DateTime.Today.Date;

        //this uses the repository I outlined above
        var foos = await _fooRepository.Select(account, effectiveDate);

        //this repository uses a rest client, I doubt it's the problem
        var bars = await _barRepository.Select(account, effectiveDate);

        //just trying to make this example less lengthy
        var foobars = MakeFoobars(foos, bars);
        var result = new ItemCreationResult { EffectiveDate = effectiveDate, Items = foobars };
        return result;
    }
}

据我所尝试：

• 使用SemaphoreSlim节流
• 没有节流
• 使用connection.OpenAnync()在回购
• 包含/排除finally块（应与无关using）

值得知道的是，foreach填充器中的循环运行了大约500次。本质上，这里有500个帐户的列表。对于每一个，它都需要执行一项长期运行的populate任务，其中涉及从我的Foo存储库中提取数据。

老实说，我不知道。我认为这可能与等待填充器中该任务列表中每个任务的异步db调用有关。对这个问题的任何见解都会很有帮助。

Answer 1

经过一番挖掘，我想我已经解决了这个问题。我认为我实际上并没有像我最初假设的那样遇到连接泄漏。据我现在的了解，通过连接池，当从代码中关闭 SQL 连接时，它实际上并没有消失——它只是作为空闲连接进入连接池。查看 SQL 中打开的连接仍然会显示它。

由于我的数据访问是异步的，因此在任何“关闭”连接返回到池之前打开的所有连接，这意味着为每个请求打开一个新连接。这导致我看到的打开连接数量惊人，让我认为我有连接泄漏。

使用 SemaphoreSlim 实际上解决了这个问题——我只是错误地实现了它。它应该像这样工作：

public override async Task<IEnumerable<ProcessResult>> ProcessAccount(DateTime? popDate = null)
{
      foreach (item in accountNumbers)
      {

      trackedTasks.Add(new Func<Task<ProcessResult>>(async () =>
            {
                await ss.WaitAsync().ConfigureAwait(false);
                try
                {
                    return await ProcessAccount(item.AccountCode, popDate ?? DateTime.Today).ConfigureAwait(false);
                }
                catch (Exception e)
                {
                    //log, etc.
                }
                finally
                {
                    ss.Release();
                }
            })());
      }
}

这样做会限制一次打开的连接数量，并等待它们关闭，因此池中相同的较小连接组将被重复使用。