Parallelen iteration in C#?

.NET 4 ist die BlockingCollection macht dies ziemlich einfach. Erstellen Sie eine BlockingCollection, wieder in seine .GetConsumingEnumerable() in der enumerable-Methode. Dann die foreach fügt einfach die Blockierung Sammlung.

E. g.

private BlockingCollection<T> m_data = new BlockingCollection<T>();

public IEnumerable<T> GetData( IEnumerable<IEnumerable<T>> sources )
{
    Task.Factory.StartNew( () => ParallelGetData( sources ) );
    return m_data.GetConsumingEnumerable();
}

private void ParallelGetData( IEnumerable<IEnumerable<T>> sources )
{
    foreach( var source in sources )
    {
        foreach( var item in source )
        {
            m_data.Add( item );
        };
    }

    //Adding complete, the enumeration can stop now
    m_data.CompleteAdding();
}

Hoffe, das hilft.
BTW ich veröffentlicht einen blog über diese Letzte Nacht

Andre

Schrieb ich eine Implementierung von EachParallel() aus der .NET4 Parallel library. Es ist kompatibel mit .NET 3.5: Die parallele ForEach-Schleife in C# 3.5
Verwendung:

string[] names = { "cartman", "stan", "kenny", "kyle" };
names.EachParallel(name =>
{
    try
    {
        Console.WriteLine(name);
    }
    catch { /* handle exception */ }
});

Umsetzung:

///<summary>
///Enumerates through each item in a list in parallel
///</summary>
public static void EachParallel<T>(this IEnumerable<T> list, Action<T> action)
{
    //enumerate the list so it can't change during execution
    list = list.ToArray();
    var count = list.Count();

    if (count == 0)
    {
        return;
    }
    else if (count == 1)
    {
        //if there's only one element, just execute it
        action(list.First());
    }
    else
    {
        //Launch each method in it's own thread
        const int MaxHandles = 64;
        for (var offset = 0; offset < list.Count() / MaxHandles; offset++)
        {
            //break up the list into 64-item chunks because of a limitiation             //in WaitHandle
            var chunk = list.Skip(offset * MaxHandles).Take(MaxHandles);

            //Initialize the reset events to keep track of completed threads
            var resetEvents = new ManualResetEvent[chunk.Count()];

            //spawn a thread for each item in the chunk
            int i = 0;
            foreach (var item in chunk)
            {
                resetEvents[i] = new ManualResetEvent(false);
                ThreadPool.QueueUserWorkItem(new WaitCallback((object data) =>
                {
                    int methodIndex = (int)((object[])data)[0];

                    //Execute the method and pass in the enumerated item
                    action((T)((object[])data)[1]);

                    //Tell the calling thread that we're done
                    resetEvents[methodIndex].Set();
                }), new object[] { i, item });
                i++;
            }

            //Wait for all threads to execute
            WaitHandle.WaitAll(resetEvents);
        }
    }
}

Wäre diese Arbeit für Sie?

public static class Parallel
{
    public static void ForEach<T>(IEnumerable<T>[] sources,
                                  Action<T> action)
    {
        foreach (var enumerable in sources)
        {
            ThreadPool.QueueUserWorkItem(source => {
                foreach (var item in (IEnumerable<T>)source)
                    action(item);
            }, enumerable);
        }
    }
}

//sample usage:
static void Main()
{
    string[] s1 = { "1", "2", "3" };
    string[] s2 = { "4", "5", "6" };
    IEnumerable<string>[] sources = { s1, s2 };
    Parallel.ForEach(sources, s => Console.WriteLine(s));
    Thread.Sleep(0); //allow background threads to work
}

Für C# 2.0, müssen Sie zum konvertieren von lambda-Ausdrücken vor die Delegierten.

Hinweis: Dieses Dienstprogramm-Methode verwendet hintergrund-threads. Möchten Sie vielleicht ändern Sie es zu einem Einsatz Vordergrund-threads, und wahrscheinlich werden Sie wollen, zu warten, bis alle threads fertig stellen. Wenn Sie das tun, ich schlage vor, Sie erstellen sources.Length - 1 threads und die aktuellen Ausführungsthread für die letzten (oder ersten) Quelle.

(Ich wünschte, ich könnte zählen waiting for threads to finish in meinem code, aber ich bin traurig, dass ich nicht weiß, wie man das doch machen. Ich denke, man sollte mit ein WaitHandle Thread.Join().)