C#从PostgreSQL批量读取15万条数据过慢的原因及优化方案

我正在C#中对PostgreSQL数据库执行批量读取操作，并非数据库专家，怀疑代码存在明显问题。尝试了两种方法读取约15万条记录，方法1耗时43秒，方法2耗时117秒。希望能在1-2秒内读取该数据，这是否现实？有没有更优的查询方法？

方法1（Dapper实现） #

private const string HeartbeatsQuery = @"SELECT id, client_id AS clientId, run_id AS runId, lat, long, speed, 
                                                     flow_rate_gallon_min AS flowRateGallonMin, cumulative_flow_pulse_count AS CumulativeFlowPulseCount, 
                                                     image, event_date_utc AS eventDateUtc, device_serial_number AS deviceSerialNumber 
                                                FROM heartbeats";

public async Task<List<Heartbeat>> GetHeartbeats(int clientId, Guid runId)
{
    var query = $@"{HeartbeatsQuery} 
                   WHERE client_id=@clientId AND run_id=@runId
                   ORDER BY event_date_utc";

    using (var connection = new NpgsqlConnection(_connectionString))
    {
        var heartbeats = await connection.QueryAsync<Heartbeat>(
            query,
            param: new
            {
                clientId = clientId,
                runId = runId
            });

        return heartbeats?.ToList() ?? new List<Heartbeat>();
    }
}

方法2（PostgreSQL COPY二进制导出） #

public async Task<List<Heartbeat>> GetHeartbeats2(int clientId, Guid runId)
{
    var heartbeats = new List<Heartbeat>();
    using (var connection = new NpgsqlConnection(_connectionString))
    {
        connection.Open();
        using (var reader = await connection.BeginBinaryExportAsync(
            @$"COPY  (SELECT client_id, run_id, lat, long, speed, flow_rate_gallon_min, cumulative_flow_pulse_count,
            image, event_date_utc, device_serial_number FROM heartbeats WHERE run_id='{runId}')
            TO STDOUT (FORMAT BINARY)"))
        {
            while (await reader.StartRowAsync() > 0)
            {
                if (reader.IsNull) await reader.SkipAsync();
                var heartbeat = new Heartbeat();

                heartbeat.ClientId = await reader.ReadAsync<int>(NpgsqlTypes.NpgsqlDbType.Integer);
                heartbeat.RunId = await reader.ReadAsync<Guid>(NpgsqlTypes.NpgsqlDbType.Uuid);

                if (reader.IsNull)
                {
                    await reader.SkipAsync();
                }
                else
                {
                    heartbeat.Lat = await reader.ReadAsync<decimal>(NpgsqlTypes.NpgsqlDbType.Numeric);
                }

                if (reader.IsNull)
                {
                    await reader.SkipAsync();
                }
                else
                {
                    heartbeat.Long = await reader.ReadAsync<decimal>(NpgsqlTypes.NpgsqlDbType.Numeric);
                }

                if (reader.IsNull)
                {
                    await reader.SkipAsync();
                }
                else
                {
                    heartbeat.Speed = await reader.ReadAsync<decimal>(NpgsqlTypes.NpgsqlDbType.Numeric);
                }

                if (reader.IsNull)
                {
                    await reader.SkipAsync();
                }
                else
                {
                    heartbeat.FlowRateGallonMin = await reader.ReadAsync<decimal>(NpgsqlTypes.NpgsqlDbType.Numeric);
                }

                if (reader.IsNull)
                {
                    await reader.SkipAsync();
                }
                else
                {
                    heartbeat.CumulativeFlowPulseCount = await reader.ReadAsync<decimal>(NpgsqlTypes.NpgsqlDbType.Bigint);
                }

                if (reader.IsNull)
                {
                    await reader.SkipAsync();
                }
                else
                {
                    heartbeat.Image = await reader.ReadAsync<string>(NpgsqlTypes.NpgsqlDbType.Text);
                }

                if (reader.IsNull)
                {
                    await reader.SkipAsync();
                }
                else
                {
                    heartbeat.EventDateUtc = await reader.ReadAsync<DateTime>(NpgsqlTypes.NpgsqlDbType.TimestampTz);
                }

                if (reader.IsNull)
                {
                    await reader.SkipAsync();
                }
                else 
                {
                    heartbeat.DeviceSerialNumber = await reader.ReadAsync<string>(NpgsqlTypes.NpgsqlDbType.Text);
                }
                
                heartbeats.Add(heartbeat);
            }
        }
    }

    return heartbeats;
}

解答：1-2秒读取完全现实，核心优化方案如下 #

一、优先解决数据库索引问题（性能瓶颈核心） #

当前耗时高的首要原因几乎可以肯定是全表扫描，给heartbeats表创建复合覆盖索引：

CREATE INDEX idx_heartbeats_client_run ON heartbeats(client_id, run_id, event_date_utc);

这个索引同时覆盖WHERE过滤条件（client_id、run_id）和排序字段（event_date_utc），让数据库无需扫描全表，直接通过索引获取数据，能把查询耗时降低一个数量级。

验证索引是否生效：执行EXPLAIN ANALYZE加上你的查询语句，查看输出是否显示使用了idx_heartbeats_client_run索引，避免Seq Scan（全表扫描）。

二、优化方法1（Dapper实现） #

Dapper本身是高效的ORM，只需做少量调整：

1. 避免字符串拼接查询：把完整查询定义为常量，让PostgreSQL缓存执行计划：

   private const string GetHeartbeatsQuery = @"SELECT id, client_id AS clientId, run_id AS runId, lat, long, speed, 
                                                flow_rate_gallon_min AS flowRateGallonMin, cumulative_flow_pulse_count AS CumulativeFlowPulseCount, 
                                                image, event_date_utc AS eventDateUtc, device_serial_number AS deviceSerialNumber 
                                            FROM heartbeats
                                            WHERE client_id=@clientId AND run_id=@runId
                                            ORDER BY event_date_utc";
   

   方法内直接使用该常量，无需拼接。
2. 确认连接池配置：Npgsql默认开启连接池，确保连接字符串中Max Pool Size设置合理（比如50），避免频繁创建销毁连接的开销。
3. 开启二进制协议：连接字符串添加UseBinaryProtocol=true（默认已开启，可确认），降低数据传输的序列化开销。

三、修复方法2的致命问题 #

COPY二进制导出理论上比普通查询更快，但你的实现有几个严重缺陷：

1. SQL注入风险+执行计划无法缓存：直接拼接runId='{runId}'，不仅不安全，还会导致PostgreSQL每次都重新生成执行计划。可以改用参数化的准备语句，或者通过临时表传递参数。
2. 逻辑不一致：方法2的WHERE条件只过滤了run_id，没加client_id，会读取更多数据，这是耗时更长的直接原因之一。
3. 逐行异步读取的开销：手动逐字段异步读取的累加开销极大，建议改用Npgsql提供的批量映射方式，或者将COPY数据导出到内存流后批量解析。

四、其他通用优化 #

• 减少数据传输量：如果image字段是大文本/二进制且本次业务不需要，直接从SELECT中移除；若必须读取，考虑延迟加载。
• 分页读取（可选）：如果业务允许分批次处理数据，可采用分页（比如每页10000条），降低单次内存占用和传输压力。
• 调整命令超时：连接字符串或Command中设置CommandTimeout=60，避免长查询超时。

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