TPL 数据流 - 为什么我的代码不使用并行处理？

Question

好的 - 首先，代码...

Imports System.Threading.Tasks
Imports System.Threading.Tasks.Dataflow

Public Class PaymentPipeLine


#Region "Properties"

    Public Property BufferBlock As BufferBlock(Of CPayment)
    Public Property TransformBlock As TransformBlock(Of CPayment, Task)
    Public Property ActionBlock As ActionBlock(Of Task)

    Public Property IncrementPayment As Action

#End Region

#Region "Constructor"
    Public Sub New(action As Action)
        Me.IncrementPayment = action
        BufferBlock = New BufferBlock(Of CPayment)
        TransformBlock = New TransformBlock(Of CPayment, Task)(Function(x As CPayment)
                                                                   Dim t = Task.Run(Sub()
                                                                                        x.process(Nothing)
                                                                                        Me.IncrementPayment.Invoke()
                                                                                    End Sub)
                                                                   Return t
                                                               End Function)
        ActionBlock = New ActionBlock(Of Task)(Sub(x As Task)
                                                   x.Wait()
                                               End Sub)
        Dim dataflowLinkOptions = New DataflowLinkOptions()
        dataflowLinkOptions.PropagateCompletion = True
        BufferBlock.LinkTo(TransformBlock, dataflowLinkOptions)
        TransformBlock.LinkTo(ActionBlock, dataflowLinkOptions)


    End Sub
#End Region

#Region "Methods"
    Public Sub Process()
        ActionBlock.Completion.Wait()
    End Sub

#End Region

End Class

现在，调用它的代码基本上将值发送到 BufferBlock，当它用完值时将其标记为完成，然后在管道上调用 Process() 方法...

我原以为这会导致并行处理付款，但我发现性能根本没有改善（代码的主要部分是对 Web 服务的 IO 调用以处理付款）。我只能得出结论，付款正在一笔一笔地通过……

如何获得它： a) 在添加付款时发送付款（即不要等待其他付款先通过），或 b) 与 a 相同，但限制说一次可能只有 5 个通过？

有什么想法吗？非常感谢您能提出任何建议...

马丁

Answer 1

我发现了问题。

我必须将ExecutionDataflowBlockOptions 对象与ActionBlock 的构造函数一起使用，并且在ExecutionDataflowBlockOptions 上我必须设置MaximumDegreeOfParallelization。