Согласно рекомендации Google , если вы используете синхронного издателя (производителя) и одного подписчика, следуйте алгоритму в коде Node JS в во второй половине страницы , чтобы гарантировать упорядочение обработки.
Точно так же, если у вас несколько издателей, вам нужно синхронизировать издателей, создав критический раздел между методом getPublishCounterValue и методом setPublishCounterValue, который побеждает многопоточный характер издателя.
Лучшее решение - следовать разделу
Порядок в конечном результате имеет значение
Типичные случаи использования: журналы, обновления состояния
blockquote>многопоточные издатели должны прикреплять отметку времени к каждому сообщению о событии pub / sub, чтобы подписчики могли сохранять сообщения о событиях в Google Cloud Datastore или Firestore как юридические лица. Отдельное задание cron процессора сообщений о событиях может извлекать объекты сообщений о событиях в отсортированном по времени метке для обеспечения порядка сообщений.
Короткий ответ, нет. foreach
работы над только одним счетным за один раз.
Однако при объединении параллели enumerables в единственную Вы можете foreach
по объединенному. Я не знаю ни о ком легком, созданном в методе выполнения этого, но следующее должно работать (хотя я не протестировал его):
public IEnumerable<TSource[]> Combine<TSource>(params object[] sources)
{
foreach(var o in sources)
{
// Choose your own exception
if(!(o is IEnumerable<TSource>)) throw new Exception();
}
var enums =
sources.Select(s => ((IEnumerable<TSource>)s).GetEnumerator())
.ToArray();
while(enums.All(e => e.MoveNext()))
{
yield return enums.Select(e => e.Current).ToArray();
}
}
Затем Вы можете foreach
по возвращенному счетному:
foreach(var v in Combine(en1, en2, en3))
{
// Remembering that v is an array of the type contained in en1,
// en2 and en3.
}
Это работало бы на Вас?
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
}
Для C# 2.0 необходимо преобразовать лямбда-выражения выше в делегатов.
Примечание: Этот служебный метод использует фоновые потоки. Можно хотеть изменить его для использования приоритетных потоков, и вероятно Вы захотите ожидать, пока все потоки не заканчиваются. Если Вы делаете это, я предлагаю, чтобы Вы создали sources.Length - 1
потоки и использование текущий поток выполнения для последнего (или сначала) источник.
(Мне бы хотелось включать ожидание потоков для окончания в моем коде, но я сожалею, что не знаю, как сделать это все же. Я предполагаю, что необходимо использовать a WaitHandle
Thread.Join()
.)
Ответ Zooba хорош, но Вы могли бы также хотеть посмотреть на ответы на, "Как выполнить итерации более чем двух массивов сразу".
Коллекция BlockingCollection .NET 4 значительно упрощает эту задачу. Создайте BlockingCollection, верните его .GetConsumingEnumerable () в методе перечисления. Затем foreach просто добавляет к блокирующей коллекции.
Например,
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();
}
Надеюсь, это поможет. Кстати, я опубликовал блог об этом вчера вечером
Андре
Я написал реализацию EachParallel () из библиотеки .NET4 Parallel. Он совместим с .NET 3.5: Параллельный цикл ForEach в C # 3.5 Использование:
string[] names = { "cartman", "stan", "kenny", "kyle" };
names.EachParallel(name =>
{
try
{
Console.WriteLine(name);
}
catch { /* handle exception */ }
});
Реализация:
/// <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);
}
}
}