多次循环遍历 DataTable Rows

Question

::编辑::

好吧，伙计们，看来我毕竟是个白痴。这个问题与我的代码完全无关，与 Visual Studio 用以前的（和空的）版本覆盖了我的 SQLite 数据库有关。似乎关于线程安全的讨论很精彩，所以我会继续阅读更多内容！

::/编辑::

我正在尝试使用多个后台工作人员一次循环遍历数据库 100 条记录的行，同时避免重复，但我似乎遇到了一些问题。 基本上，我首先在一个循环中创建 10 个后台工作人员，然后将它们添加到一个列表中。然后我循环遍历列表中的后台工作人员，并为每个工作人员执行 RunWorkerAsync()，然后将主线程睡眠 5 秒。在每个后台工作人员的 DoWork 方法中，我让工作人员从数据库中选择 100 行，其中特定字段设置为其默认值。 从这里开始，我想首先遍历每个返回的行并将默认值更改为“进行中”值，然后再次遍历这些值并实际执行为这些字段找到正确值所需的处理。我似乎遇到的问题是，在第一次迭代结果后我似乎有一个空的 DataTable，我怀疑我的问题源于浅拷贝。 代码如下：

using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.IO;
using System.Text;
using DBFill.GeoCodeService;
using System.Diagnostics;
using System.Runtime.Serialization.Formatters.Binary;

namespace DBFill {
    class Program {
        public static int completedGeocodes = 0;
        static void Main(string[] args) {
            SQLiteDatabase db = new SQLiteDatabase("zipCodes.s3db");
            List<BackgroundWorker> workers = new List<BackgroundWorker>();
            for (int i = 0; i < 10; i++) {
                BackgroundWorker b = new BackgroundWorker();
                b.DoWork += new DoWorkEventHandler(worker_DoWork);
                b.RunWorkerCompleted += new RunWorkerCompletedEventHandler(worker_RunWorkerCompleted);
                b.WorkerReportsProgress = true;
                b.ProgressChanged += new ProgressChangedEventHandler(b_ProgressChanged);
                workers.Add(b);
            }
            int counter = 0;
            foreach (BackgroundWorker b in workers) {
                Debug.WriteLine("Worker {0} is starting.", counter);
                b.RunWorkerAsync(b);
                counter++;
                System.Threading.Thread.Sleep(5000);
            }
            Boolean running = true;
            while (running) {
                running = false;
                foreach (BackgroundWorker b in workers) {
                    Debug.WriteLine("Checking background Worker");
                    if (b.IsBusy) {
                        running = true;
                    }
                }
                System.Threading.Thread.Sleep(5000);
            }

        }

        static void b_ProgressChanged(object sender, ProgressChangedEventArgs e) {
            Console.WriteLine(".");
        }

        static void worker_RunWorkerCompleted(object sender, RunWorkerCompletedEventArgs e) {

        }

        static void worker_DoWork(object sender, DoWorkEventArgs e) {
            BackgroundWorker b = (BackgroundWorker)e.Argument;
            SQLiteDatabase db = new SQLiteDatabase("zipCodes.s3db");
            DataTable results = get100Records();
            DataTable temp = DeepClone<DataTable>(results);//results;
            Dictionary<String, String> marker = new Dictionary<string, string>();
            marker["LATITUDE"] = "In Progress";
            foreach (DataRow row in temp.Rows) {
                Debug.WriteLine("Working with zip {0}", row["ZIP_CODE"]);
                db.Update("ZIP_CODES", marker, String.Format("ZIP_CODE = '{0}'", row["ZIP_CODE"]));
            }
            foreach (DataRow row in results.Rows) {
                String geoCodeResponse = GeoCodeZip(row["ZIP_CODE"].ToString());
                Debug.WriteLine(String.Format("Attempting Zip: {0}", row["ZIP_CODE"].ToString()));
                if (geoCodeResponse != "There was an error") {
                    marker["LATITUDE"] = geoCodeResponse.Split(',')[0];
                    marker["LONGITUDE"] = geoCodeResponse.Split(',')[1];
                    Console.WriteLine(String.Format("#{0} updated successfully", completedGeocodes));
                }
                else {
                    marker["LATITUDE"] = "Not Set";
                    Console.WriteLine(String.Format("#{0} failed", completedGeocodes));
                }
                db.Update("ZIP_CODES", marker, String.Format("ZIP_CODE = '{0}'", row["ZIP_CODE"]));
                db.ExecuteNonQuery("commit");
                b.ReportProgress(1);
                completedGeocodes++;
            }
        }

        private static DataTable get100Records() {
            SQLiteDatabase db = new SQLiteDatabase("zipCodes.s3db");
            DataTable results = db.GetDataTable("select ZIP_CODE from ZIP_CODES where LATITUDE = 'Not Set' LIMIT 100");
            return results;
        }

        private static String GeoCodeZip(String zip) {
            try {
                GeocodeRequest request = new GeocodeRequest();
                request.Credentials = new GeoCodeService.Credentials();
                request.Credentials.ApplicationId = "API_KEY";
                request.Query = zip;
                ConfidenceFilter[] filters = new ConfidenceFilter[1];
                filters[0] = new ConfidenceFilter();
                filters[0].MinimumConfidence = Confidence.High;
                GeocodeOptions opts = new GeocodeOptions();
                opts.Filters = filters;
                request.Options = opts;
                GeocodeServiceClient service = new GeocodeServiceClient("BasicHttpBinding_IGeocodeService");
                GeocodeResponse response = service.Geocode(request);
                if (response.Results.Length > 0) {
                    return String.Format("{0},{1}", response.Results[0].Locations[0].Latitude, response.Results[0].Locations[0].Longitude);
                }
                else {
                    Debug.WriteLine(String.Format("{0}", response.ResponseSummary.FaultReason));
                    return "There was an error";
                }
            }
            catch (Exception e) {
                Debug.WriteLine(e.Message);
                return "There was an error";
            }
        }

        public static T DeepClone<T>(T obj) {
            using (var ms = new MemoryStream()) {
                var formatter = new BinaryFormatter();
                formatter.Serialize(ms, obj);
                ms.Position = 0;

                return (T)formatter.Deserialize(ms);
            }
        }

    }
}

有什么想法吗？

Answer 1

看来您的延迟以及为什么要对其进行多线程处理，不是从数据库中读取记录，而是调用GeocodeServiceClient。

您可以尝试修改您的 main 方法以按顺序从数据库中获取所有记录并解析它们。然后，您将该列表拆分为多个块，并启动后台工作人员以通过地理编码服务运行它们。

另一种选择是将记录放入队列中，让每个后台工作人员从队列中弹出一个处理它，然后在仍有未处理的记录时返回队列。您需要小心锁定In C# would it be better to use Queue.Synchronized or lock() for thread safety?。