.net - 如何使用连接 PLINQ 现有的 LINQ 查询？

Question

我正在使用 LINQ 将两个 DataSet 相互比较以创建新行并更新现有行。我注意到完整的比较持续了大约 1.5 个小时，并且两个核心中只有一个是忙碌的(任务管理器是 50-52% 的 CPU 使用率)。我必须承认，我对并行 LINQ 完全陌生，但我认为它可以显着提高性能。

所以我的问题是，我应该如何并行化以及并行化什么？

这些是原始查询(减少到基本要素):

'check for new data
Dim srcUnique = From row In src.Email_Total
                Select Ticket_ID = row.ticket_id, Interaction = row.interaction, ModifiedAt = row.modified_time

Dim destUnique = From row In dest.ContactDetail
                 Where row.ContactRow.fiContactType = emailContactType.idContactType
                 Select row.ContactRow.Ticket_ID, row.Interaction, row.ModifiedAt

'get all emails(contactdetails) that are in source but not in destination
Dim diffRows = srcUnique.Except(destUnique).ToList

'get all new emails(according to ticket_id) for calculating contact columns
Dim newRowsTickets = (From row In src.Email_Total
                     Join d In diffRows
                     On row.ticket_id Equals d.Ticket_ID _
                     And row.interaction Equals d.Interaction _
                     And row.modified_time Equals d.ModifiedAt
                     Group row By Ticket_ID = row.ticket_id Into NewTicketRows = Group).ToList

For Each ticket In newRowsTickets
     Dim contact = dest.Contact.FindByTicket_IDfiContactType(ticket.Ticket_ID, emailContactType.idContactType)
     If contact Is Nothing Then
          ' Create new Contact with many sub-queries on this ticket(omitted) ****'
          Dim newContact = Me.dest.Contact.NewContactRow
          dest.Contact.AddContactRow(newContact)
          contact = newContact
     Else
          ' Update Contact with many sub-queries on this ticket(omitted) '
     End If
     daContact.Update(dest.Contact)

     ' Add new ContactDetail-Rows from this Ticket(this is the counterpart of the src.Email_Total-Rows, details omitted) '
     For Each newRow In ticket.NewTicketRows
         Dim newContactDetail = dest.ContactDetail.NewContactDetailRow
         newContactDetail.ContactRow = contact
         dest.ContactDetail.AddContactDetailRow(newContactDetail)
     Next
     daContactDetails.Update(dest.ContactDetail)
Next

备注 : daContact和 daContactDetails是 SqlDataAdapters , source和 dest是 DataSets和 Contact和 ContactDetail是 DataTables ，其中每个 ContactDetail 都属于一个联系人。

即使不是两个核心都使用 100% CPU，我认为如果我将查询并行化，它会显着提高性能，因为第二个核心几乎空闲。 for each也可能是优化的好地方，因为票证彼此不相关。所以我假设我可以用多个线程循环并并行创建/更新记录。但是如何用 PLINQ 做到这一点呢？

旁注 :正如我在评论中提到的，到目前为止，性能对我来说并不是一个关键因素，因为服务器的唯一目的是将 MySQL 数据库(在另一台服务器上)与 MS SQL 服务器(在与此相同的服务器上)同步Windows-服务)。它充当由其他服务生成的报告的来源。但是这些报告每天只生成一次。但除此之外，我对学习 PLINQ 很感兴趣，因为我认为这可能是一个很好的练习。
仅当目标 DB 为空且必须创建所有记录时才需要上述 1,5h。如果两个数据库几乎同步，此方法只需要大约 1 分钟。在 future ，性能将变得更加重要，因为电子邮件只是几种联系类型之一(聊天+通话将超过 100 万条记录)。我认为无论如何我都需要某种(LINQ)数据分页。

如果有什么不清楚的，我会相应地更新我的答案。提前致谢。

编辑 :这是我的调查和尝试的结果:

问题:如何使用连接“PLINQ”现有的 LINQ 查询？

Answer :请注意，某些 LINQ 运算符是二进制的——它们将两个 IEnumerables 作为输入。 Join 是此类运算符的完美示例。在这些情况下，最左侧数据源的类型决定了使用 LINQ 还是 PLINQ。因此，您只需在第一个数据源上调用 AsParallel 即可并行运行查询:

IEnumerable<T> leftData = ..., rightData = ...;
var q = from x in leftData.AsParallel()
        join y in rightData on x.a == y.b
        select f(x, y);

但是，如果我按以下方式更改查询(请注意 AsParallel ):

Dim newRowsTickets = (From row In src.Email_Total.AsParallel()
                                        Join d In diffRows
                                        On row.ticket_id Equals d.Ticket_ID _
                                        And row.interaction Equals d.Interaction _
                                        And row.modified_time Equals d.ModifiedAt
                                    Group row By Ticket_ID = row.ticket_id Into NewTicketRows = Group).ToList

编译器会提示我需要添加 AsParallel到正确的数据源。所以这似乎是 VB.NET 问题或缺乏文档(文章来自 2007)。我假设是后者，因为(除了那篇推荐的)文章还说你需要添加 System.Concurrency.dll手动但实际上它是 .NET 4.0 Framework 的一部分，并且在命名空间中 Sytem.Threading.Tasks .

我意识到我不会从并行化 Except 中获利由于查询在顺序模式下足够快(即使两个集合中的行数几乎相同，这导致了最大比较次数，我在不到 30 秒的时间内就得到了结果)。但为了完整起见，我稍后会添加它。

所以我决定并行化 for-each使用 LINQ-Queries 很简单，您只需要添加 AsParallel()最后。
但我意识到我需要强制与 WithExecutionMode(ParallelExecutionMode.ForceParallelism) 并行。 , 否则 .NET 决定在此循环中仅使用一个核心。我还想告诉 .NET 我希望使用尽可能多的线程但不超过 8 个: WithDegreeOfParallelism(8).
现在两个内核同时工作，但 CPU 使用率保持在 54%。

所以这是到目前为止的 PLINQ 版本:

Dim diffRows = srcUnique.AsParallel.Except(destUnique.AsParallel).ToList

Dim newRowsTickets = (From row In src.Email_Total.AsParallel()
                        Join d In diffRows.AsParallel()
                        On row.ticket_id Equals d.Ticket_ID _
                        And row.interaction Equals d.Interaction _
                        And row.modified_time Equals d.ModifiedAt
                    Group row By Ticket_ID = row.ticket_id Into NewTicketRows = Group).ToList

For Each ticket In newRowsTickets.
                    AsParallel().
                      WithDegreeOfParallelism(8).
                       WithExecutionMode(ParallelExecutionMode.ForceParallelism)
    '  blah,blah ...  '

    'add new ContactDetails for this Ticket(only new rows)
    For Each newRow In ticket.NewTicketRows.
                                AsParallel().
                                    WithExecutionMode(ParallelExecutionMode.Default)
        ' blah,blah ... '
    Next
    daContactDetails.Update(dest.ContactDetail)
Next

不幸的是，我没有看到使用 AsParallel 带来的任何性能优势。与顺序模式相比:
for each与 AsParallel (时:分:秒.毫米):

09/29/2011 18:54:36: Contacts/ContactDetails created or modified. Duration: 01:21:34.40

并且没有:

09/29/2011 16:02:55: Contacts/ContactDetails created or modified. Duration: 01:21:24.50

有人可以解释我这个结果吗？ for each 中的数据库是否具有写访问权限？负责类似的时间？

以下是推荐读物:

http://msdn.microsoft.com/en-us/magazine/cc163329.aspx (不是最新的)

List of changes since above article

Overview of videos/articles according this topic

PDF: When To Use ParallelForEach Or PLINQ

When PLINQ will speedup performance

Answer 1

有3点值得进一步研究，

不要使用 .toList()。我可能错了，但我认为使用 .ToList
这种方式不允许编译器优化查询，如果
进一步优化是可能的。

使用您自己的过滤操作来比较两者的数据
目的地。它可能会给你更好的性能。

看看能不能用LinqDataview提供更好的
性能。

我认为您在插入时不会获得 PLinq 的优势。看 this answer了解更多详情。

希望有帮助。请务必询问您是否需要澄清上述任何一点。