用微服务替代BCP工具：解决大数据量查询内存溢出问题

你的核心问题是一次性将大量数据加载到内存导致溢出，BCP之所以高效是因为它采用流式处理，不会把所有数据存到内存里。下面分语言给出最优的流式导出方案，同时兼顾并发和未来更大数据量的需求：

C# (ASP.NET Core 推荐方案) #

利用ASP.NET Core的响应流式能力，结合SqlDataReader的SequentialAccess模式，逐行读取并写入响应流，全程不把数据缓存到内存中：

[HttpGet("export-employees")]
public async Task<IActionResult> ExportEmployees(int runId)
{
    // 设置响应头，标记为可下载的CSV文件
    Response.Headers.Add("Content-Disposition", $"attachment; filename=employees_{runId}.csv");
    Response.ContentType = "text/csv";

    await using var connection = new SqlConnection(Configuration.GetConnectionString("DefaultConnection"));
    await connection.OpenAsync();

    // 避免SELECT *，只查询需要的列能大幅减少数据量
    await using var command = new SqlCommand("SELECT Id, Name, Department, HireDate FROM Employees WHERE runid = @RunId", connection);
    command.Parameters.AddWithValue("@RunId", runId);
    
    // SequentialAccess让DataReader按列顺序读取，不预加载整行到内存，适合大数据场景
    await using var reader = await command.ExecuteReaderAsync(CommandBehavior.SequentialAccess);

    var streamWriter = new StreamWriter(Response.Body);
    
    // 写入CSV表头
    var columnNames = Enumerable.Range(0, reader.FieldCount).Select(reader.GetName);
    await streamWriter.WriteLineAsync(string.Join(",", columnNames.Select(c => $"\"{c.Replace("\"", "\"\"")}\"")));

    // 逐行读取并写入响应流
    while (await reader.ReadAsync())
    {
        var rowValues = Enumerable.Range(0, reader.FieldCount)
            .Select(i => 
            {
                var value = reader.IsDBNull(i) ? string.Empty : reader.GetValue(i).ToString();
                // CSV转义规则：双引号替换为两个双引号
                return $"\"{value.Replace("\"", "\"\"")}\"";
            });
        await streamWriter.WriteLineAsync(string.Join(",", rowValues));
        // 及时刷新缓冲区，避免内存堆积
        await streamWriter.FlushAsync();
    }

    await streamWriter.FlushAsync();
    return new EmptyResult();
}

额外优化 #

• 启用响应压缩：在Program.cs中添加builder.Services.AddResponseCompression()和app.UseResponseCompression()，减少网络传输量
• 配置Kestrel：调整MaxRequestBodySize、MaxConcurrentConnections参数，适配高并发请求
• 数据库连接池：在连接字符串中配置Max Pool Size，避免并发时连接耗尽

Java (Spring Boot) #

使用JdbcTemplate的流式查询能力，直接将数据写入HttpServletResponse的输出流：

@GetMapping("/export-employees")
public void exportEmployees(@RequestParam int runId, HttpServletResponse response) throws IOException, SQLException {
    response.setContentType("text/csv");
    response.setHeader("Content-Disposition", "attachment; filename=employees_" + runId + ".csv");

    try (PrintWriter writer = response.getWriter();
         Connection conn = DriverManager.getConnection("jdbc:sqlserver://localhost:1433;databaseName=YourDb;user=sa;password=YourPass")) {

        String sql = "SELECT Id, Name, Department FROM Employees WHERE runid = ?";
        try (PreparedStatement stmt = conn.prepareStatement(sql)) {
            stmt.setInt(1, runId);
            try (ResultSet rs = stmt.executeQuery()) {
                ResultSetMetaData metaData = rs.getMetaData();
                int columnCount = metaData.getColumnCount();

                // 写入表头
                for (int i = 1; i <= columnCount; i++) {
                    writer.print("\"" + metaData.getColumnName(i).replace("\"", "\"\"") + "\"");
                    if (i < columnCount) writer.print(",");
                }
                writer.println();

                // 逐行写入数据
                while (rs.next()) {
                    for (int i = 1; i <= columnCount; i++) {
                        String value = rs.getString(i);
                        value = value == null ? "" : value.replace("\"", "\"\"");
                        writer.print("\"" + value + "\"");
                        if (i < columnCount) writer.print(",");
                    }
                    writer.println();
                    writer.flush();
                }
            }
        }
    }
}

Python (FastAPI) #

利用FastAPI的Response实现流式输出，配合pyodbc的逐行查询：

from fastapi import FastAPI, Response
import pyodbc
from typing import Iterator

app = FastAPI()

def generate_csv(run_id: int) -> Iterator[str]:
    conn_str = "DRIVER={ODBC Driver 17 for SQL Server};SERVER=localhost;DATABASE=YourDb;UID=sa;PWD=YourPass"
    conn = pyodbc.connect(conn_str)
    cursor = conn.cursor()
    cursor.execute("SELECT Id, Name, Department FROM Employees WHERE runid = ?", (run_id,))
    
    # 生成表头
    column_names = [col[0] for col in cursor.description]
    yield ','.join([f'"{name.replace("\"", "\"\"")}"' for name in column_names]) + '\n'
    
    # 逐行生成数据行
    for row in cursor:
        row_values = []
        for val in row:
            str_val = str(val) if val is not None else ''
            row_values.append(f'"{str_val.replace("\"", "\"\"")}"')
        yield ','.join(row_values) + '\n'
    
    cursor.close()
    conn.close()

@app.get("/export-employees")
async def export_employees(run_id: int, response: Response):
    response.headers["Content-Disposition"] = f"attachment; filename=employees_{run_id}.csv"
    response.headers["Content-Type"] = "text/csv"
    return Response(content=generate_csv(run_id), media_type="text/csv")

Node.js (Express) #

使用mssql库的流式查询功能，直接将数据写入响应流：

const express = require('express');
const sql = require('mssql');
const app = express();

const dbConfig = {
    user: 'sa',
    password: 'YourPass',
    server: 'localhost',
    database: 'YourDb',
    options: { encrypt: false }
};

app.get('/export-employees', async (req, res) => {
    const runId = parseInt(req.query.runId);
    res.setHeader('Content-Disposition', `attachment; filename=employees_${runId}.csv`);
    res.setHeader('Content-Type', 'text/csv');

    try {
        await sql.connect(dbConfig);
        const request = new sql.Request();
        request.input('RunId', sql.Int, runId);
        request.stream = true; // 启用流式查询

        request.query('SELECT Id, Name, Department FROM Employees WHERE runid = @RunId');

        // 处理元数据，写入表头
        request.on('metadata', (columns) => {
            const header = columns.map(col => `"${col.name.replace(/"/g, '""')}"`).join(',');
            res.write(header + '\n');
        });

        // 处理每行数据
        request.on('row', (row) => {
            const rowValues = Object.values(row).map(val => {
                const strVal = val === null ? '' : String(val);
                return `"${strVal.replace(/"/g, '""')}"`;
            }).join(',');
            res.write(rowValues + '\n');
        });

        request.on('done', () => {
            res.end();
            sql.close();
        });

        request.on('error', (err) => {
            console.error(err);
            res.status(500).end('导出失败');
            sql.close();
        });
    } catch (err) {
        console.error(err);
        res.status(500).end('数据库连接失败');
    }
});

app.listen(3000, () => console.log('服务启动在3000端口'));

通用优化准则 #

1. 避免SELECT*：只查询业务需要的列，减少数据传输量和内存占用
2. 数据库索引优化：为runid字段创建索引，提升查询效率
3. 并发控制：配置数据库连接池大小，避免并发请求耗尽连接；调整Web服务器的并发连接数限制
4. 监控与告警：添加内存、CPU、数据库连接数的监控，及时发现性能瓶颈

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