如何用C#高效向MySQL批量导入100万条数据并减少数据库调用

核心优化方向 #

• 摒弃逐条执行UPDATE/INSERT的逻辑，改用批量操作将多次数据库调用合并为少数几次
• 用事务包裹所有批量操作，保证数据一致性的同时减少事务提交开销
• 分批次处理数据（比如每批10000条），避免内存占用过高

可用技术/库 #

1. MySQL Connector/NET（官方驱动） #

官方原生支持批量操作，性能最优，无需额外依赖。

2. Dapper（轻量ORM） #

在原生驱动基础上简化代码，支持批量SQL拼接或参数化批量操作，兼顾性能与开发效率。

3. Entity Framework Core（EF Core） #

适合已有EF栈的项目，通过AddRange+SaveChanges或原生SQL批量操作实现。

代码示例 #

示例1：MySQL Connector/NET 批量更新+插入 #

using MySqlConnector;
using System.Collections.Generic;
using System.Data;

public class BatchDataProcessor
{
    private readonly string _connectionString;

    public BatchDataProcessor(string connectionString)
    {
        _connectionString = connectionString;
    }

    public void ProcessBatch(List<UpdateStatusEntity> entities)
    {
        // 分批次，每批10000条
        int batchSize = 10000;
        for (int i = 0; i < entities.Count; i += batchSize)
        {
            var batch = entities.Skip(i).Take(batchSize).ToList();
            using var conn = new MySqlConnection(_connectionString);
            conn.Open();
            using var transaction = conn.BeginTransaction();

            try
            {
                // 批量更新tblatr
                UpdateTblAtrBatch(conn, transaction, batch);
                // 批量插入tblatrhistory
                InsertTblAtrHistoryBatch(conn, transaction, batch);

                transaction.Commit();
            }
            catch
            {
                transaction.Rollback();
                throw;
            }
        }
    }

    private void UpdateTblAtrBatch(MySqlConnection conn, MySqlTransaction transaction, List<UpdateStatusEntity> batch)
    {
        // 假设UpdateStatusEntity有Id、Status、UpdateTime字段，Id是tblatr主键
        var updateSql = @"
            UPDATE tblatr 
            SET Status = CASE Id 
                {0}
            END,
            UpdateTime = CASE Id
                {1}
            END
            WHERE Id IN ({2})";

        var statusCases = new List<string>();
        var timeCases = new List<string>();
        var ids = new List<string>();
        var parameters = new List<MySqlParameter>();

        foreach (var entity in batch)
        {
            string paramStatusName = $"@status_{entity.Id}";
            string paramTimeName = $"@time_{entity.Id}";
            statusCases.Add($"WHEN {entity.Id} THEN {paramStatusName}");
            timeCases.Add($"WHEN {entity.Id} THEN {paramTimeName}");
            ids.Add(entity.Id.ToString());
            parameters.Add(new MySqlParameter(paramStatusName, entity.Status));
            parameters.Add(new MySqlParameter(paramTimeName, entity.UpdateTime));
        }

        string finalSql = string.Format(updateSql, 
            string.Join(" ", statusCases), 
            string.Join(" ", timeCases), 
            string.Join(",", ids));

        using var cmd = new MySqlCommand(finalSql, conn, transaction);
        cmd.Parameters.AddRange(parameters.ToArray());
        cmd.ExecuteNonQuery();
    }

    private void InsertTblAtrHistoryBatch(MySqlConnection conn, MySqlTransaction transaction, List<UpdateStatusEntity> batch)
    {
        var insertSql = @"
            INSERT INTO tblatrhistory (AtrId, Status, CreateTime)
            VALUES {0}";

        var valueClauses = new List<string>();
        var parameters = new List<MySqlParameter>();

        foreach (var entity in batch)
        {
            string paramAtrId = $"@hist_atrid_{entity.Id}";
            string paramStatus = $"@hist_status_{entity.Id}";
            string paramCreateTime = $"@hist_createtime_{entity.Id}";
            
            valueClauses.Add($"({paramAtrId}, {paramStatus}, {paramCreateTime})");
            parameters.Add(new MySqlParameter(paramAtrId, entity.Id));
            parameters.Add(new MySqlParameter(paramStatus, entity.Status));
            parameters.Add(new MySqlParameter(paramCreateTime, entity.UpdateTime));
        }

        string finalSql = string.Format(insertSql, string.Join(",", valueClauses));

        using var cmd = new MySqlCommand(finalSql, conn, transaction);
        cmd.Parameters.AddRange(parameters.ToArray());
        cmd.ExecuteNonQuery();
    }
}

// 实体类示例
public class UpdateStatusEntity
{
    public int Id { get; set; }
    public string Status { get; set; }
    public DateTime UpdateTime { get; set; }
}

示例2：Dapper 批量操作简化版 #

using Dapper;
using MySqlConnector;
using System.Collections.Generic;

public class DapperBatchProcessor
{
    private readonly string _connectionString;

    public DapperBatchProcessor(string connectionString)
    {
        _connectionString = connectionString;
    }

    public void ProcessBatch(List<UpdateStatusEntity> entities)
    {
        int batchSize = 10000;
        for (int i = 0; i < entities.Count; i += batchSize)
        {
            var batch = entities.Skip(i).Take(batchSize).ToList();
            using var conn = new MySqlConnection(_connectionString);
            conn.Open();
            using var transaction = conn.BeginTransaction();

            try
            {
                // 批量更新
                var updateSql = @"
                    UPDATE tblatr 
                    SET Status = @Status, UpdateTime = @UpdateTime
                    WHERE Id = @Id";
                conn.Execute(updateSql, batch, transaction);

                // 批量插入历史表
                var insertSql = @"
                    INSERT INTO tblatrhistory (AtrId, Status, CreateTime)
                    VALUES (@Id, @Status, @UpdateTime)";
                conn.Execute(insertSql, batch, transaction);

                transaction.Commit();
            }
            catch
            {
                transaction.Rollback();
                throw;
            }
        }
    }
}

额外性能优化建议 #

• 临时禁用非必要索引：在批量操作前关闭tblatr和tblatrhistory的非主键索引，操作完成后重建，避免频繁更新索引的开销
• 关闭自动提交：默认MySQL是自动提交事务，手动开启事务并批量提交能大幅减少IO
• 使用参数化查询：避免SQL注入，同时让MySQL缓存查询计划，提升重复执行效率
• 调整MySQL配置：增大max_allowed_packet（避免批量SQL过大报错）、调整innodb_buffer_pool_size提升缓存能力

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