使用生产者/消费者模式和 SqlBulkCopy 到 SQL Server DB 中使用多线程处理块中的平面文件

Question

我希望你能容忍我。我想提供尽可能多的信息。 主要问题是如何创建一个由多个线程使用的结构（如堆栈），该结构将弹出一个值并使用它来处理一个大的平面文件，并且可能会一次又一次地循环，直到处理整个文件。 如果文件有 100.000 条记录，可以由 5 个线程使用 2.000 行块处理 那么每个线程将得到 10 个块来处理。

我的目标是在平面文件中移动数据（带有标题...子标题...详细信息，详细信息，详细信息，...详细信息，子页脚，子标题...详细信息，详细信息，详细信息，...详细信息，子页脚， Subheader...详细信息、详细信息、详细信息、...详细信息、子页脚、页脚结构）到 OLTP 数据库中，该数据库的恢复模式为简单（可能完整）到 3 个表中：第一个代表子标题行中存在的子标题的唯一键，第二个是中间表表 SubheaderGroup，表示 2000 条记录块中的详细信息行分组（需要将 Subheader 的 Identity PK 作为其 FK，第三个表示详细信息行，其中 FK 指向 Subheader PK。

我正在进行手动事务管理，因为我可以拥有数以万计的详细信息行 我在加载期间使用目标表中设置为 0 的特殊字段，然后在文件处理结束时我正在执行事务性更新，将此值更改为 1，这可以向其他应用程序发出加载完成的信号。

我想将此平面文件分成多个相等的部分（相同的行数），这些部分可以使用多个线程进行处理，并使用 SqlBulkCopy 使用从目标表元数据创建的 IDataReader 进行导入。

我想使用生产者/消费者模式（如下面的链接 - pdf 分析和代码示例中所述）将 SqlBulkCopy 与 SqlBulkCopyOptions.TableLock 选项一起使用。 http://sqlblog.com/blogs/ alberto_ferrari/archive/2009/11/30/sqlbulkcopy-performance-analysis.aspx 此模式允许创建多个生产者，并且需要订阅生产者以消耗该行的同等数量的消费者。

在TestSqlBulkCopy项目中，DataProducer.cs文件中有一个方法可以模拟生成数千条记录。

public void Produce (DataConsumer consumer, int numberOfRows) {
    int bufferSize = 100000;
    int numberOfBuffers = numberOfRows / bufferSize;

    for (int bufferNumber = 0; bufferNumber < numberOfBuffers; bufferNumber++) {
        DataTable buffer = consumer.GetBufferDataTable ();

        for (int rowNumber = 0; rowNumber < bufferSize; rowNumber++) {
            object[] values = GetRandomRow (consumer);
            buffer.Rows.Add (values);
        }
        consumer.AddBufferDataTable (buffer);
    }
}

该方法将在新线程的上下文中执行。我希望这个新线程只读取原始平面文件的一个唯一块，另一个线程将开始处理下一个块。然后，消费者将使用 SqlBulkCopy ADO.NET 类将数据（泵送至他们）移动到 SQL Server DB。

所以这里的问题是关于主程序规定每个线程应该处理什么 lineFrom 到 lineTo ，我认为这应该在线程创建期间发生。 第二种解决方案可能是让线程共享一些结构并使用它们特有的东西（例如线程号或序列号）来查找共享结构（可能是一个堆栈并弹出一个值（在执行此操作时锁定堆栈），然后下一个线程将然后拾取下一个值。主程序将选取平面文件并确定块的大小并创建堆栈。

因此，有人可以提供一些代码片段，即有关多个线程如何处理一个文件并仅获取唯一部分的代码片段。那个文件

？ 拉德

Answer 1

对我来说效果好的方法是使用队列来保存未处理的工作，并使用字典来跟踪正在进行的工作：

1. 创建一个工作类，它接受
   文件名、起始行和行数
   并且有一个更新方法
   数据库是否插入。传递一个回调方法
   工作人员用来在完成时发出信号。
2. 加载带有工作线程实例的队列
   类，每个块一个。
3. 生成一个调度程序线程，使
   工作实例，启动其更新
   方法，并将工作实例添加到字典中，并以其线程的 ManagedThreadId 为键。这样做
   直到你允许的最大线程数
   达到计数，如所示
   词典.计数.调度员
   等待直到线程完成
   然后启动另一个。有几种方法可以让它等待。
4. 当每个线程完成时，它的回调
   从中删除其 ManagedThreadId
   字典。如果线程退出
   由于错误（例如
   连接超时）然后
   回调可以重新插入worker
   进入队列。这是个好地方
   更新您的用户界面。
5. 您的 UI 可以显示活动线程、总进度和每个块的时间。它可以让用户调整活动线程的数量、暂停处理、显示错误或提前停止。
6. 当队列和字典为空时，您就完成了。

作为控制台应用程序的演示代码：

using System;
using System.Collections.Generic;
using System.Threading;

namespace threadtest
{
    public delegate void DoneCallbackDelegate(int idArg, bool successArg, string messageArg);

    class Program
    {
        static void Main(string[] args)
        {
            Supervisor supv = new Supervisor();
            supv.LoadQueue();
            supv.Dispatch();
        }
    }

    public class Supervisor
    {
        public Queue<Worker> pendingWork = new Queue<Worker>();
        public Dictionary<int, Worker> activeWork = new Dictionary<int, Worker>();

        private object pendingLock = new object();
        private object activeLock = new object();

        private int maxThreads = 200;

        public void LoadQueue()
        {
            for (int i = 0; i < 1000; i++)
            {
                Worker worker = new Worker();
                worker.Callback = new DoneCallbackDelegate(WorkerFinished);
                lock (pendingLock)
                {
                    pendingWork.Enqueue(worker);
                }
            }
        }

        public void Dispatch()
        {
            int activeThreadCount;

            while (true)
            {
                lock (activeLock) { activeThreadCount = activeWork.Count; }
                while (true)
                {
                    lock (activeLock)
                    {
                        if (activeWork.Count == maxThreads) break;
                    }
                    lock (pendingWork)
                    {
                        if (pendingWork.Count > 0)
                        {
                            Worker worker = pendingWork.Dequeue();
                            Thread thread = new Thread(new ThreadStart(worker.DoWork));
                            thread.IsBackground = true;
                            worker.ThreadId = thread.ManagedThreadId;
                            lock (activeLock) { activeWork.Add(worker.ThreadId, worker); }
                            thread.Start();
                        }
                        else
                        {
                            break;
                        }
                    }
                }
                Thread.Sleep(200); // wait to see if any workers are done (many ways to do this)

                lock (pendingLock)
                    lock (activeLock)
                    {
                        if ((pendingWork.Count == 0) && (activeWork.Count == 0)) break;
                    }
            }
        }

        // remove finished threads from activeWork, resubmit if necessary, and update UI
        public void WorkerFinished(int idArg, bool successArg, string messageArg)
        {
            lock (pendingLock)
                lock (activeLock)
                {
                    Worker worker = activeWork[idArg];
                    activeWork.Remove(idArg);
                    if (!successArg)
                    {
                        // check the message or something to see if you should resubmit thread
                        pendingWork.Enqueue(worker);
                    }
                    // update UI
                    int left = Console.CursorLeft;
                    int top = Console.CursorTop;
                    Console.WriteLine(string.Format("pending:{0} active:{1}        ", pendingWork.Count, activeWork.Count));
                    Console.SetCursorPosition(left, top);
                }
        }
    }

    public class Worker
    {
        // this is where you put in your problem-unique stuff
        public int ThreadId { get; set; }

        DoneCallbackDelegate callback;
        public DoneCallbackDelegate Callback { set { callback = value; } }

        public void DoWork()
        {
            try
            {
                Thread.Sleep(new Random().Next(500, 5000)); // simulate some effort
                callback(ThreadId, true, null);
            }
            catch (Exception ex)
            {
                callback(ThreadId, false, ex.ToString());
            }
        }
    }
}