异步流式处理大XML写入Oracle CLOB的类库及代码优化咨询

询问是否存在可异步流式处理大型XML并存储至Oracle CLOB列的类库。目前正在开发通用的XML流式读写至Oracle XML CLOB的功能，多方检索未找到适配类库，现有代码需优化——将每次读取插入新记录改为更新CLOB列，以下是两种实现代码及并行化方案：

现有实现代码1 #

static async Task Main()
{
    const string connectionString = "<your connection string>";
    const string tableName = "<your table name>";
    const int chunkSize = 100000;

    using var connection = new OracleConnection(connectionString);
    await connection.OpenAsync();
    using var transaction = connection.BeginTransaction();
    using var command = connection.CreateCommand();
    command.Transaction = transaction;
    command.CommandText = $"INSERT INTO {tableName} (xml_clob) VALUES (:clob)";
    var parameter = new OracleParameter("clob", OracleDbType.Clob);

    const string filePath = "large-file.xml";
    using var reader = XmlReader.Create(filePath);
    var semaphore = new SemaphoreSlim(50);

    var tasks = new List<Task<int>>();
    var settings = new XmlWriterSettings
    {
        Indent = true,
        IndentChars = "    "
    };

    while (await reader.ReadAsync())
    {
        switch (reader.NodeType)
        {
            case XmlNodeType.Element:
                using var ms = new MemoryStream();
                using var writer = XmlWriter.Create(ms, settings);
                writer.WriteStartElement(reader.Name);
                if (reader.HasAttributes)
                {
                    while (reader.MoveToNextAttribute())
                    {
                        writer.WriteAttributeString(reader.Name, reader.Value);
                    }

                    reader.MoveToElement();
                }
                writer.Flush();
                ms.Seek(0, SeekOrigin.Begin);
                parameter.Value = new OracleClob(connection, true);
                using var clobWriter = new StreamWriter(parameter.Value, Encoding.Unicode);
                await ms.CopyToAsync(clobWriter.BaseStream);
                await clobWriter.FlushAsync();
                var task = command.ExecuteNonQueryAsync();
                tasks.Add(task);
                break;

            case XmlNodeType.EndElement:
                break;

            case XmlNodeType.Text:
                using var ms2 = new MemoryStream();
                using var writer2 = XmlWriter.Create(ms2);
                while (!reader.EOF)
                {
                    char[] buffer = new char[chunkSize];
                    int charsRead = await reader.ReadValueChunkAsync(buffer, 0, buffer.Length);

                    if (charsRead == 0)
                    {
                        break;
                    }

                    await writer2.WriteCharsAsync(buffer, 0, charsRead);
                }

                writer2.Flush();
                ms2.Seek(0, SeekOrigin.Begin);
                parameter.Value = new OracleClob(connection, true);
                using var clobWriter2 = new StreamWriter(parameter.Value, Encoding.Unicode);
                await ms2.CopyToAsync(clobWriter2.BaseStream);
                await clobWriter2.FlushAsync();
                var task2 = command.ExecuteNonQueryAsync();
                tasks.Add(task2);
                break;
        }
    }

    await Task.WhenAll(tasks);
    await transaction.CommitAsync();
    Console.WriteLine($"Finished processing XML file '{filePath}'");
}

并行化提升写入速度的方案 #

using Oracle.ManagedDataAccess.Client;
using System;
using System.IO;
using System.Threading.Tasks;

public async Task SaveFileToOracleClobParallel(string filePath, string connectionString)
{
    const int bufferSize = 81920; // 80KB buffer size
    const int maxDegreeOfParallelism = 4; // maximum number of tasks to run in parallel

    try
    {
        using (var connection = new OracleConnection(connectionString))
        {
            await connection.OpenAsync();

            using (var transaction = connection.BeginTransaction())
            using (var command = connection.CreateCommand())
            {
                command.Transaction = transaction;
                command.CommandText = "INSERT INTO MyTable (MyClobColumn) VALUES (:clob)";

                using (var clob = new OracleClob(connection))
                {
                    await clob.OpenAsync(OracleDbType.Clob, System.Data.ConnectionState.Open);

                    var tasks = new List<Task>();

                    using (var stream = new FileStream(filePath, FileMode.Open, FileAccess.Read, FileShare.Read, bufferSize, FileOptions.SequentialScan))
                    {
                        for (int i = 0; i < maxDegreeOfParallelism; i++)
                        {
                            tasks.Add(ReadAndWriteChunkAsync(stream, clob, bufferSize));
                        }

                        await Task.WhenAll(tasks);
                    }

                    await clob.CloseAsync();
                    command.Parameters.Add(new OracleParameter("clob", OracleDbType.Clob) { Value = clob });

                    await command.ExecuteNonQueryAsync();
                    await transaction.CommitAsync();
                }
            }
        }
    }
    catch (Exception ex)
    {
        Console.WriteLine($"Error: {ex.Message}");
        // Handle the exception as appropriate
    }
}

private async Task ReadAndWriteChunkAsync(Stream stream, OracleClob clob, int bufferSize)
{
    var buffer = new byte[bufferSize];
    int bytesRead;

    while ((bytesRead = await stream.ReadAsync(buffer, 0, buffer.Length)) > 0)
    {
        await clob.WriteAsync(buffer, 0, bytesRead);
    }
}

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