你的初步思路方向是对的，但从性能、可靠性和可维护性角度，我给你补充一些优化点和具体的实现代码——毕竟处理百万级数据，细节没做好很容易出问题（比如内存爆掉、重复导出或者数据丢失）。

先确认下你的流程逻辑没问题，但可以做这几个关键调整：

• 用流式读取代替一次性把10万条数据加载到内存，减少GC压力（SqlDataReader就是干这个的，别用DataSet一次性拉所有数据）
• 标记已处理的操作要和数据读取做原子性保障，比如在存储过程里用UPDATE ... OUTPUT直接返回要处理的记录——这样能避免多个进程/线程同时处理时的重复数据问题
• 文件写入用批量缓冲，别写一条刷一次磁盘，大幅提升IO效率

下面是完整的可参考代码，包含数据读取、文件拆分和状态更新的核心逻辑：

1. 数据库操作辅助方法 #

首先封装数据库的读取和更新逻辑，确保资源自动释放：

using System.Data.SqlClient;
using System.IO;
using System.Collections.Generic;
using System.Linq;

public class DataExporter
{
    private readonly string _connectionString;

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

    // 读取一批未处理的记录（用OUTPUT子句原子性标记+返回数据）
    public IEnumerable<YourRecordModel> GetNextBatch(int batchSize = 100000)
    {
        using var conn = new SqlConnection(_connectionString);
        conn.Open();
        using var cmd = new SqlCommand("GetNextUnprocessedBatch", conn);
        cmd.CommandType = System.Data.CommandType.StoredProcedure;
        cmd.Parameters.Add(new SqlParameter("@BatchSize", batchSize));

        using var reader = cmd.ExecuteReader();
        while (reader.Read())
        {
            // 映射到你的实体模型，根据实际字段调整
            yield return new YourRecordModel
            {
                Id = reader.GetInt32(0),
                Column1 = reader.GetString(1),
                Column2 = reader.GetDateTime(2),
                // 其他业务字段...
            };
        }
    }

    // 批量标记记录为已处理
    public void MarkRecordsAsProcessed(IEnumerable<int> recordIds)
    {
        using var conn = new SqlConnection(_connectionString);
        conn.Open();
        using var cmd = new SqlCommand("MarkRecordsAsProcessed", conn);
        cmd.CommandType = System.Data.CommandType.StoredProcedure;

        // 用表值参数传递批量ID，比循环调用高效N倍
        var tvp = new DataTable();
        tvp.Columns.Add("Id", typeof(int));
        foreach (var id in recordIds)
        {
            tvp.Rows.Add(id);
        }
        cmd.Parameters.Add(new SqlParameter("@RecordIds", tvp) { TypeName = "dbo.IdListType" });

        cmd.ExecuteNonQuery();
    }
}

// 你的业务实体模型，根据数据库表结构定义
public class YourRecordModel
{
    public int Id { get; set; }
    public string Column1 { get; set; }
    public DateTime Column2 { get; set; }
    // 其他业务字段...
}

2. 文件写入逻辑 #

实现按批次写入Flat文件，注意缓冲和格式处理：

public void ExportToFlatFiles(string outputDirectory, int batchSize = 100000)
{
    if (!Directory.Exists(outputDirectory))
    {
        Directory.CreateDirectory(outputDirectory);
    }

    int fileCounter = 1;
    while (true)
    {
        // 读取一批数据
        var batch = GetNextBatch(batchSize).ToList();
        if (!batch.Any())
        {
            break; // 没有更多数据，退出循环
        }

        // 生成带序号的文件名，比如 DataBatch_001.txt
        string filePath = Path.Combine(outputDirectory, $"DataBatch_{fileCounter:D3}.txt");
        // 1MB缓冲的StreamWriter，减少磁盘IO次数
        using var writer = new StreamWriter(filePath, false, Encoding.UTF8, 1024 * 1024);

        // 写入表头（如果不需要可以注释掉）
        writer.WriteLine("Id,Column1,Column2");

        // 批量写入记录，注意处理特殊字符避免格式错误
        foreach (var record in batch)
        {
            // 处理CSV格式的引号转义：把"替换为""
            string escapedColumn1 = record.Column1.Replace("\"", "\"\"");
            string line = $"{record.Id},\"{escapedColumn1}\",{record.Column2:yyyy-MM-dd HH:mm:ss}";
            writer.WriteLine(line);
        }

        // 写完文件再标记为已处理，避免文件写入失败但数据被误标记
        MarkRecordsAsProcessed(batch.Select(r => r.Id));

        Console.WriteLine($"已完成文件 {filePath}，共 {batch.Count} 条记录");
        fileCounter++;
    }

    Console.WriteLine("所有数据导出完成！");
}

3. SQL Server存储过程示例 #

对应的存储过程，用UPDATE ... OUTPUT实现原子性操作，彻底避免并发重复读取：

-- 先创建表值参数类型，用于批量传递ID
CREATE TYPE dbo.IdListType AS TABLE (Id INT);
GO

-- 获取下一批未处理记录的存储过程
CREATE PROCEDURE GetNextUnprocessedBatch
    @BatchSize INT
AS
BEGIN
    SET NOCOUNT ON;
    BEGIN TRANSACTION;

    -- 原子性标记为处理中并返回数据，避免并发冲突
    UPDATE TOP(@BatchSize) YourTableName
    SET Status = 'Processing'
    OUTPUT inserted.Id, inserted.Column1, inserted.Column2 -- 返回需要导出的字段
    WHERE Status = 'Unprocessed';

    COMMIT TRANSACTION;
END
GO

-- 标记记录为已处理的存储过程
CREATE PROCEDURE MarkRecordsAsProcessed
    @RecordIds dbo.IdListType READONLY
AS
BEGIN
    SET NOCOUNT ON;
    UPDATE YourTableName
    SET Status = 'Processed'
    WHERE Id IN (SELECT Id FROM @RecordIds);
END
GO

• 原子性保障：用UPDATE ... OUTPUT的方式，能确保同一批数据不会被多个进程读取，就算程序崩溃，标记为Processing的记录也可以手动改回Unprocessed重新处理
• 内存优化：如果数据量特别大，甚至可以边读边写，不用把整批数据加载到List里——只需要在读取时收集ID，写完文件再更新状态
• 错误处理：一定要加try-catch块，比如文件写入失败时，要把标记为Processing的记录改回Unprocessed，避免数据丢失
• 性能调优：给Status字段加非聚集索引，加快存储过程的查询速度；根据磁盘IO性能调整StreamWriter的缓冲大小；可以用异步方法（SqlDataReader.ReadAsync、StreamWriter.WriteLineAsync）提升吞吐量
• 日志记录：每批处理完成后记录日志，方便排查异常问题

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