使用Pomelo+Dapper时QueryAsync出现间歇性MySQL连接错误求助

问题现象 #

频繁出现以下间歇性数据库连接错误，难以复现但日志中出现频率高：

This MySqlConnection is already in use.
Connection must be Open; current state is Connecting

使用Pomelo.EntityFrameworkCore.MySql作为EF Core数据库提供方，通过Dapper的QueryAsync执行查询操作。

相关配置与代码 #

连接字符串 #

Server=****;DataBase=*****;Uid=***;Pwd=****;default command timeout=0;SslMode=none;max pool size=1000;Connect Timeout=300;convert zero datetime=True;ConnectionIdleTimeout=5;Pooling=true;MinimumPoolSize=25

核心查询代码 #

var query = $@"
    SELECT 
    at.*, dpt.*, c.*, cnt.*, mEnc.*, atUsr.*, usr.*
    FROM Atendimento AS at
    INNER JOIN Departamento AS dpt ON at.DepartamentoId = dpt.Id
    INNER JOIN Canal AS c ON at.CanalId = c.Id
    LEFT JOIN Contato AS cnt ON at.ContatoId = cnt.Id
    LEFT JOIN MotivoEncerramento as mEnc ON at.MotivoEncerramentoId = mEnc.id
    LEFT JOIN (
    AtendimentoUsuario AS atUsr
    INNER JOIN Users AS usr ON atUsr.UserId = usr.Id
    ) ON at.Id = atUsr.AtendimentoId
    WHERE at.IdRef = '{idRef}'
    ORDER BY dpt.Id, c.Id, atUsr.Id, usr.Id;";

var atendimentos = await Dapper.SqlMapper.QueryAsync<Atendimento, Departamento, Canal, Contato, MotivoEncerramento, AtendimentoUsuario, Usuario, Atendimento>(
    _dbContext.Database.GetDbConnection(),
    query,
    (atendimento, departamento, canal, contato, motivoEncerramento, atUsuario, usuario) =>
    {
        atendimento.Departamento = departamento;
        atendimento.Contato = contato;
        atendimento.MotivoEncerramento = motivoEncerramento;
        atendimento.Canal = canal;
        atendimento.AtendimentoUsuarios = new List<AtendimentoUsuario>();
        if (atUsuario is not null)
        {
            atUsuario.Usuario = usuario;
            atendimento.AtendimentoUsuarios.Add(atUsuario);
        }

        return atendimento;
    });

var result = atendimentos.GroupBy(a => a.Id).Select(g =>
{
    var groupedAtendimento = g.First();
    if (g.Any(c => c.AtendimentoUsuarios.Count > 0))
    {
        groupedAtendimento.AtendimentoUsuarios = g.Select(a => a.AtendimentoUsuarios.SingleOrDefault()).ToList();
    }

    return groupedAtendimento;
});

atendimento = result.First();

EF Core连接获取方法 #

//
// Summary:
//     Gets the underlying ADO.NET System.Data.Common.DbConnection for this Microsoft.EntityFrameworkCore.DbContext.
//     This connection should not be disposed if it was created by Entity Framework.
//     Connections are created by Entity Framework when a connection string rather than
//     a DbConnection object is passed to the 'UseMyProvider' method for the database
//     provider in use. Conversely, the application is responsible for disposing a DbConnection
//     passed to Entity Framework in 'UseMyProvider'.
//
// Parameters:
//   databaseFacade:
//     The Microsoft.EntityFrameworkCore.Infrastructure.DatabaseFacade for the context.
//
// Returns:
//     The System.Data.Common.DbConnection
//
// Remarks:
//     See Connections and connection strings for more information.
public static DbConnection GetDbConnection(this DatabaseFacade databaseFacade)
{
    return GetFacadeDependencies(databaseFacade).RelationalConnection.DbConnection;
}

解决方案建议 #

1. 停止复用EF Core管理的DbConnection
   EF Core的_dbContext.Database.GetDbConnection()返回的连接由EF内部维护，可能处于被EF占用的状态（如Connecting、In Use），直接传给Dapper会引发并发冲突。正确做法是让Dapper自行管理连接：

   using var connection = new MySqlConnection(_connectionString);
   await connection.OpenAsync();
   var atendimentos = await Dapper.SqlMapper.QueryAsync<Atendimento, Departamento, Canal, Contato, MotivoEncerramento, AtendimentoUsuario, Usuario, Atendimento>(
       connection,
       query,
       (atendimento, departamento, canal, contato, motivoEncerramento, atUsuario, usuario) =>
       {
           // 映射逻辑保持不变
           atendimento.Departamento = departamento;
           atendimento.Contato = contato;
           atendimento.MotivoEncerramento = motivoEncerramento;
           atendimento.Canal = canal;
           atendimento.AtendimentoUsuarios = new List<AtendimentoUsuario>();
           if (atUsuario is not null)
           {
               atUsuario.Usuario = usuario;
               atendimento.AtendimentoUsuarios.Add(atUsuario);
           }
   
           return atendimento;
       });
   

2. 立即加载Dapper查询结果到内存
   Dapper的QueryAsync返回延迟枚举的结果，后续的GroupBy、First操作才会触发实际数据库查询，此时连接状态可能已变化。必须在查询后调用ToListAsync()提前完成数据库操作：

   var atendimentos = await Dapper.SqlMapper.QueryAsync<...>(...).ToListAsync();
   

3. 优化连接池配置
   当前max pool size=1000过大、ConnectionIdleTimeout=5过短，易导致连接频繁创建销毁，增加冲突概率：

   • 将max pool size调整为100-200（根据业务并发量合理设置）
   • 延长ConnectionIdleTimeout至30-60秒
   • 给default command timeout设置合理值（建议300秒而非0，避免连接长时间占用）

4. 修复SQL注入风险
   代码中使用字符串拼接生成SQL，存在严重注入问题，必须改用参数化查询：

   var query = @"
       SELECT at.*, dpt.*, c.*, cnt.*, mEnc.*, atUsr.*, usr.*
       FROM Atendimento AS at
       -- 其余JOIN逻辑不变
       WHERE at.IdRef = @idRef
       ORDER BY dpt.Id, c.Id, atUsr.Id, usr.Id;";
   
   var atendimentos = await Dapper.SqlMapper.QueryAsync<...>(connection, query, new { idRef }, ...);
   

内容的提问来源于stack exchange，提问作者Rafael Ferrato