c# - Guid.NewGuid() 在 Parallel.For 循环中使用时返回重复值

Question

我有一个正在访问 API 的应用程序。因此，它对对象中的所有 ID 进行查询，然后必须对每个 ID 一次查询一个项目。我在 Parallel.For 循环中执行此操作并将每个项目数据添加到数据表中的一行。然后我使用 sqlbulkcopy 将数据表发送到 SQL 服务器表。

如果我在不使用 Parallel.For 的情况下执行此操作，则效果很好。但是，对于 Parallel.For，这一行:

workrow["id"] = Guid.NewGuid();

正在生成重复的 Guid。它经常这样做并导致数据无法加载到 SQL 服务器表中，因为 SQL 中的 id 行是主键并且不允许重复。我尝试锁定:

                    lock (lockobject)
                    {
                        workrow["id"] = Guid.NewGuid();
                    }

这没有帮助。
我尝试不为该字段分配一个 id，希望 SQL 会生成它(它在该字段上确实有 newid())。说它不能插入空值失败了。
我似乎不能只是从数据表中删除 id 字段，因为当我执行 sqlbulkcopy 时，列不会对齐。

有人可以在这里帮助我吗？我要么需要弄清楚如何让 Guid.NewGuid() 停止生成重复项，要么我需要找出一种不传入 id(始终是数据表中的第一个字段)的方法，以便 SQL 生成 id。

这是我用来生成表之一的代码:

        public static DataTable MakeWorkflowTable()
        {
            DataTable Workflow = new DataTable("Workflow");
            DataColumn id = new DataColumn("id", System.Type.GetType("System.Guid"));
            Workflow.Columns.Add(id);
            DataColumn OrgInfoID = new DataColumn("OrgInfoID", System.Type.GetType("System.Guid"));
            Workflow.Columns.Add(OrgInfoID);
            DataColumn Name = new DataColumn("Name", System.Type.GetType("System.String"));
            Workflow.Columns.Add(Name);
            DataColumn Active = new DataColumn("Active", System.Type.GetType("System.String"));
            Workflow.Columns.Add(Active);
            DataColumn Description = new DataColumn("Description", System.Type.GetType("System.String"));
            Workflow.Columns.Add(Description);
            DataColumn Object = new DataColumn("Object", System.Type.GetType("System.String"));
            Workflow.Columns.Add(Object);
            DataColumn Formula = new DataColumn("Formula", System.Type.GetType("System.String"));
            Workflow.Columns.Add(Formula);
            DataColumn ManageableState = new DataColumn("ManageableState", System.Type.GetType("System.String"));
            Workflow.Columns.Add(ManageableState);
            DataColumn NameSpacePrefix = new DataColumn("NameSpacePrefix", System.Type.GetType("System.String"));
            Workflow.Columns.Add(NameSpacePrefix);
            DataColumn TDACount = new DataColumn("TDACount", System.Type.GetType("System.Int32"));
            Workflow.Columns.Add(TDACount);
            DataColumn TriggerType = new DataColumn("TriggerType", System.Type.GetType("System.String"));
            Workflow.Columns.Add(TriggerType);
            DataColumn CreatedDate = new DataColumn("CreatedDate", System.Type.GetType("System.DateTime"));
            Workflow.Columns.Add(CreatedDate);
            DataColumn CreatedBy = new DataColumn("CreatedBy", System.Type.GetType("System.String"));
            Workflow.Columns.Add(CreatedBy);
            DataColumn LastModifiedDate = new DataColumn("LastModifiedDate", System.Type.GetType("System.DateTime"));
            Workflow.Columns.Add(LastModifiedDate);
            DataColumn LastModifiedBy = new DataColumn("LastModifiedBy", System.Type.GetType("System.String"));
            Workflow.Columns.Add(LastModifiedBy);
            return Workflow;
        }

这是我用来将其发送到 SQL 服务器的代码:

        public static void SendDTtoDB(ref DataTable dt, ref SqlConnection cnn, string TableName)
        {
            using (SqlBulkCopy bulkCopy = new SqlBulkCopy(cnn))
            {
                bulkCopy.DestinationTableName =
                    TableName;
                try
                {
                    bulkCopy.WriteToServer(dt);
                    dt.Clear();
                }
                catch (Exception e)
                {
                    logger.Warn("SendDTtoDB {TableName}: ORGID: {ORGID} : {Message}", TableName, dt.Rows[0]["OrgInfoID"], e.Message.ToString());
                    if (e.Message.ToString().Contains("PRIMARY KEY"))
                    {
                        foreach(DataRow row in dt.Rows)
                        {
                            logger.Warn("ID: {id}", row["id"]);
                        }
                    }
                }
            }

        }

正如您在 catch 语句中看到的那样，我将它设置为将 ID 写到日志中，这样我就可以自己查看它们，果然，那里有一个重复项。太令人沮丧了!如果不需要的话，我真的不想取出 Parallel.For 和单线程。

每个请求，这里是 Parallel.For 的代码

              if (qr.totalSize > 0)
                {
                    object lockobject = new object();
                    Parallel.For(0, qr.records.Length, i =>
                    {
                        ToolingService.CustomTab1 vr = new ToolingService.CustomTab1();

                        vr = (ToolingService.CustomTab1)qr.records[i];
                        string mdSOQL = "Select FullName, description, ManageableState, MasterLabel, NamespacePrefix, Type, Url, CreatedDate, CreatedBy.Name, "
                            + "LastModifiedDate, LastModifiedBy.Name From CustomTab where id='" + vr.Id + "'";
                        ToolingService.QueryResult mdqr = new ToolingService.QueryResult();
                        ToolingService.CustomTab1 vrmd = new ToolingService.CustomTab1();
                        mdqr = ts.query(mdSOQL);
                        vrmd = (ToolingService.CustomTab1)mdqr.records[0];

                        DataRow workrow = CustomTabs.NewRow();
                        lock (lockobject)
                        {
                            workrow["id"] = Guid.NewGuid();
                        }
                        workrow["OrgInfoID"] = _orgDBID;
                        workrow["FullName"] = vrmd.FullName;
                        workrow["Description"] = vrmd.Description ?? Convert.DBNull;
                        workrow["ManageableState"] = vrmd.ManageableState;
                        workrow["MasterLabel"] = vrmd.MasterLabel ?? Convert.DBNull;
                        workrow["NameSpacePrefix"] = vrmd.NamespacePrefix ?? Convert.DBNull;
                        workrow["Type"] = vrmd.Type ?? Convert.DBNull;
                        workrow["URL"] = vrmd.Url ?? Convert.DBNull;
                        workrow["CreatedDate"] = vrmd.CreatedDate ?? Convert.DBNull;
                        if (vrmd.CreatedBy == null)
                        {
                            workrow["CreatedBy"] = Convert.DBNull;
                        }
                        else
                        {
                            workrow["CreatedBy"] = vrmd.CreatedBy.Name;
                        }
                        workrow["LastModifiedDate"] = vrmd.LastModifiedDate ?? Convert.DBNull;
                        if (vrmd.LastModifiedBy == null)
                        {
                            workrow["LastModifiedBy"] = Convert.DBNull;
                        }
                        else
                        {
                            workrow["LastModifiedBy"] = vrmd.LastModifiedBy.Name;
                        }
                        lock (CustomTabs)
                        {
                            CustomTabs.Rows.Add(workrow);
                        }

                    });
                    OrgTables.SendDTtoDB(ref CustomTabs, ref _cnn, "OrgCustomTabs");

Answer 1

事实是，必须在数据表的lock中使用Parallel.ForEach，这与首先使用Parallel.ForEach的目的背道而驰；但是，我很惊讶当您调用 DataRow workrow = CustomTabs.NewRow(); 时没有收到异常，因为在我的测试中，我收到了索引损坏的异常。我实际上必须将对NewRow的调用包装在一个锁中。像这样的东西:

Parallel.ForEach(data, x =>
            {
                DataRow row = null;
                lock (lockRow)
                {
                    row = dt.NewRow();
                    row["Guid"] = Guid.NewGuid();
                }
...
                lock(lockObj)
                   dt.Rows.Add(row);

其中 lockObj和 lockRow是2个独立的静态对象，实例化为

static  object lockObj = new  object();
static  object lockRow = new object();

这对我有用，向 DataTable 添加了 100 万行并确保所有 Guid 都是唯一的。

综上所述，我强烈建议按照Julian的建议编写代码，或创建一个实现 IDataReader(可与SQLBulkCopy一起使用)的类，然后使用该类上传数据。