C# 并行扩展 Task.Factory.StartNew 在错误的对象上调用方法

Question

好的，开始使用 System.Threading.Tasks 中的 .Net 4.0 Parellel Extensions。我发现了一些看起来很奇怪的行为，但我认为我只是做错了什么。我有一个接口(interface)和几个实现类，它们对此很简单。

interface IParallelPipe
{
    void Process(ref BlockingCollection<Stream> stream, long stageId);
}

class A:IParallelPipe
{
    public void Process(ref BlockingCollection<Stream> stream, long stageId)
    {
        //do stuff
    }
}

class B:IParallelPipe
{
    public void Process(ref BlockingCollection<Stream> stream, long stageId)
    {
        //do stuff
    }
}

然后我的类(class)从这些开始。这就是问题所在。我本质上是从传入的类型中获取有关要调用什么实现类的信息，然后调用工厂来实例化它，然后我用它创建一个任务并启动它。此处显示:

BlockingCollection<Stream> bcs = new BlockingCollection<Stream>();                   
foreach (Stage s in pipeline.Stages) 
{
    IParallelPipe p = (IParallelPipe)Factory.GetPipe(s.type);
    Task.Factory.StartNew(() => p.Process(ref bcs, s.id)); 
}

在我的示例中的每次运行中，pipeline.Stages 包含两个元素，一个被实例化为 A 类，另一个被实例化为 B 类。这很好，我在 te 调试器中看到它随着 p 与两个元素一起消失不同种类。但是，永远不会调用 B 类，而是调用 A.Process(...) 方法两次。两者都包含传入的 stageId(即，两次调用具有不同的 stageId)。

现在，如果我把东西分开一点，只是为了测试，我可以通过做这样的事情来让事情工作:

BlockingCollection<Stream> bcs = new BlockingCollection<Stream>();                   
A a = null;
B b = null;
foreach (Stage s in pipeline.Stages) 
{
    IParallelPipe p = (IParallelPipe)Factory.GetPipe(s.type);
    if(p is A)
        a = p;
    else
        b = p;
}
Task.Factory.StartNew(() => a.Process(ref bcs, idThatINeed)); 
Task.Factory.StartNew(() => b.Process(ref bcs, idThatINeed));

这会调用适当的类!

有什么想法吗？？？

Answer 1

您描述的行为对我来说似乎很奇怪 - 我希望使用正确的实例，但可能使用错误的阶段 ID - old foreach variable capture问题。正在捕获变量 s，当任务工厂评估闭包时，s 的值已更改。

这肯定是您的代码中的一个问题，但它并不能解释您为什么会遇到问题。只是检查一下，您真的是在循环内而不是在循环外声明 p 吗？如果您在循环外声明 p，那将解释一切。

下面是捕获问题的解决方法:

BlockingCollection<Stream> bcs = new BlockingCollection<Stream>();
foreach (Stage s in pipeline.Stages) 
{
    Stage copy = s;
    IParallelPipe p = (IParallelPipe)Factory.GetPipe(s.type);
    Task.Factory.StartNew(() => p.Process(ref bcs, copy.id)); 
}

请注意，我们只是在循环内获取一个副本，并捕获该副本，以便每次都获得变量的不同“实例”。

或者，我们可以只捕获 ID，而不是捕获舞台，因为这就是我们所需要的:

BlockingCollection<Stream> bcs = new BlockingCollection<Stream>();
foreach (Stage s in pipeline.Stages) 
{
    long id = s.id;
    IParallelPipe p = (IParallelPipe)Factory.GetPipe(s.type);
    Task.Factory.StartNew(() => p.Process(ref bcs, id)); 
}

如果这没有帮助，您能否发布一个简短但完整的程序来演示问题？这将使追踪变得容易得多。