在多线程编程中，MapReduce 是最为常用的一种模型。在C#中有TaskFactory类，可以实现此功能。本文就通过一个实例，介绍具体的工作原理。同时，还引入了CancellationTokenSource类，用于中止整个MapReduce任务。
基本原理MapReduce 分为 Map 和 Reduce两个过程，在Map中进行并行计算，在Reduce中对并行计算的结果进行汇总。这两步在TaskFactory中，分别可以使可以有以下实现方法：
- Map factory.StartNew(task, source.Token)
需要注意的是，在调用此方法后，代码就会开始并行执行，并返回任务的句柄（Task对象）； - Reduce factory.ContinueWhenAll(tasks, reduceTask, token);
此方法会等所有任务都结束后才会执行 reduceTask 任务，同时由于 token 的引入，如果任何一个任务取消了，就会引发 AggregateException 异常。
本示例演示了使用10个线程分别生成10个[0, 199] 的随机整型（共100个），一旦在随机生成时值为0，则会调用 source.Cancel() 函数。在完成所有任务后，求所有值的平均值。为了演示一定的时间变化，中间加入了一些延时函数。
// 代码比较长，但只有两个部分：变量定义与初始化 和 任务执行
// 来源：https://docs.microsoft.com/en-us/dotnet/api/system.threading.cancellationtokensource?view=net-6.0
public void MapReduceExample()
{
// Define the cancellation token.
var source = new CancellationTokenSource();
List> tasks = new List>();
TaskFactory factory = new TaskFactory(source.Token);
Random random = new Random();
tbLogs.Text = "";
// Map Task
for (int taskId = 0; taskId < 10; taskId++)
{
Func fi = () =>
{
print($"Mapper: Task start.");
Thread.Sleep(100);
int[] values = new int[10];
for (int i = 0; i < values.Length; i++)
{
values[i] = random.Next(0, 200);
if (values[i] == 0 && !source.Token.IsCancellationRequested)
{
// 如果这里取消了，那么当前的任务就会结束，使用聚类方法调用时会触发AggregateException 异常
source.Cancel();
print($"Mapper: Cancelling.");
}
}
print($"Mapper: Task Done.");
return values;
};
// 添加任务到tasks中，实际上这里任务已经启动了 StartNew 就会启动一个新线程。
tasks.Add(factory.StartNew(fi, source.Token));
}
Thread.Sleep(1);
print("Main: tasks.Count" + tasks.Count);
// Reduce Task
Func[], double> func = (results) =>
{
print($"Reducer: Calculating overall mean...");
return results.Sum(r => r.Result.Sum()) / (double)results.Sum(r => r.Result.Length);
};
try
{
// 此时 tasks 中的线程已经都在运行中了，如果中间时间过长的话，会有部分任务已经完成了
print($"Main: Begin to calculate...");
Task fTask = factory.ContinueWhenAll(tasks.ToArray(), func, source.Token);
print($"Main: The mean is {fTask.Result}.");
}
catch (AggregateException ae)
{
print($"Main: ae = " + ae.Message);
foreach (Exception e in ae.InnerExceptions)
{
if (e is TaskCanceledException tcx)
print($"Main: Unable to compute mean: {tcx.Message}");
else
print($"Main: Exception: " + e.GetType().Name);
}
}
finally
{
source.Dispose();
}
void print(string msg)
{
var newstr = $"{DateTime.Now:HH:mm:ss.fff}-{Thread.CurrentThread.ManagedThreadId}: {msg}n";
this.Dispatcher.BeginInvoke(() => { tbLogs.Text += newstr; });
Debug.WriteLine(newstr);
}
运行结果
-
正常结果：返回分步和统计的结果。
-
运行取消：当随机值为0时，会取消整个操作。
如果不是等所有的Reducer结束，而是只要有一个就结束，可以调用 ContinueWhenAny 方法。不过这些由于使用了 CancellationTokenSource，所以会引发 ObjectDisposedException 异常，代码如下所示：
// 代码比较长，但只有两个部分：变量定义与初始化 和 任务执行
// 来源：https://docs.microsoft.com/en-us/dotnet/api/system.threading.cancellationtokensource?view=net-6.0
public void MapReduceExample()
{
// Define the cancellation token.
var source = new CancellationTokenSource();
List> tasks = new List>();
TaskFactory factory = new TaskFactory(source.Token);
Random random = new Random();
tbLogs.Text = "";
// Map Task
for (int taskId = 0; taskId < 10; taskId++)
{
Func fi = () =>
{
print($"Mapper: Task start.");
Thread.Sleep(100);
int[] values = new int[10];
for (int i = 0; i < values.Length; i++)
{
values[i] = random.Next(0, 200);
try
{
if (values[i] == 0 && !source.Token.IsCancellationRequested)
{
source.Cancel();
print($"Mapper: Cancelling.");
}
}
catch (ObjectDisposedException ex) { print("ObjectDisposedException: " + ex.Message); }
catch (Exception ex) { print(ex.ToString()); }
}
print($"Mapper: Task Done.");
return values;
};
tasks.Add(factory.StartNew(fi, source.Token));
}
Thread.Sleep(1);
print("Main: tasks.Count" + tasks.Count);
// Reduce Task
Func[], double> func = (results) =>
{
print($"Reducer: Calculating overall mean...");
return results.Sum(r => r.Result.Sum()) / (double)results.Sum(r => r.Result.Length);
};
try
{
print($"[Main]: Begin to calculate...");
// Task fTask = factory.ContinueWhenAll(tasks.ToArray(), func, source.Token);
Task fTask = factory.ContinueWhenAny(tasks.ToArray(), (res) => { return (double)res.Result.Sum(); }, source.Token);
print($"[Main]: The mean is {fTask.Result}.");
}
catch (AggregateException ae)
{
print($"Main: ae = " + ae.Message);
foreach (Exception e in ae.InnerExceptions)
{
if (e is TaskCanceledException tcx)
print($"Main: Unable to compute mean: {tcx.Message}");
else
print($"Main: Exception: " + e.GetType().Name);
}
}
finally
{
source.Dispose();
}
void print(string msg)
{
var newstr = $"{DateTime.Now:HH:mm:ss.fff}-{Thread.CurrentThread.ManagedThreadId}: {msg}n";
this.Dispatcher.BeginInvoke(() => { tbLogs.Text += newstr; });
Debug.WriteLine(newstr);
}
}
其他
PS. 在测试中出现以下代码，原因未知，暂时未解决。
详细信息如下
System.Resources.MissingSatelliteAssemblyException 未能找到或无法加载用于回退区域性“en”的名为“Microsoft.VisualStudio.DesignTools.SurfaceDesigner.resources.dll, Version=17.0.0.0, PublicKeyToken=b03f5f7f11d5a3a”的附属程序集。这通常是安装问题。请考虑重新安装或修复该应用程序。 在 System.Resources.ManifestBasedResourceGroveler.HandleSatelliteMissing() 在 System.Resources.ManifestBasedResourceGroveler.GrovelForResourceSet(CultureInfo culture, Dictionary`2 localResourceSets, Boolean tryParents, Boolean createIfNotExists, StackCrawlMark& stackMark) 在 System.Resources.ResourceManager.InternalGetResourceSet(CultureInfo requestedCulture, Boolean createIfNotExists, Boolean tryParents, StackCrawlMark& stackMark) 在 System.Resources.ResourceManager.InternalGetResourceSet(CultureInfo culture, Boolean createIfNotExists, Boolean tryParents) 在 System.Resources.ResourceManager.GetString(String name, CultureInfo culture) 在 System.Linq.Enumerable.ToDictionary[TSource,TKey,TElement](IEnumerable`1 source, Func`2 keySelector, Func`2 elementSelector, IEqualityComparer`1 comparer) 在 Microsoft.VisualStudio.DesignTools.SurfaceDesigner.DesignerTapStringTable.GetStringTable[T]() 在 Microsoft.VisualStudio.DesignTools.WpfSurfaceDesigner.Views.WpfSurfaceProcessContext.GetDesignerTapStringTable() 在 Microsoft.VisualStudio.DesignTools.SurfaceDesigner.Documents.SurfaceIsolation.SurfaceProcessContext.InitializePipeline() 在 Microsoft.VisualStudio.DesignTools.SurfaceDesigner.Documents.SurfaceIsolation.SurfaceProcessContext.FinishSurfaceProcessCreation(DateTime start, Boolean forcePlatformOnly) 在 Microsoft.VisualStudio.DesignTools.SurfaceDesigner.Documents.SurfaceIsolation.SurfaceProcessContext.d__105.MoveNext() --- 引发异常的上一位置中堆栈跟踪的末尾 --- 在 System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw() 在 System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task) 在 Microsoft.VisualStudio.DesignTools.XamlSurfaceDesigner.Views.IsolatedSurfaceImageHost. d__67.MoveNext() --- 引发异常的上一位置中堆栈跟踪的末尾 --- 在 System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw() 在 Microsoft.VisualStudio.Telemetry.WindowsErrorReporting.WatsonReport.GetClrWatsonExceptionInfo(Exception exceptionObject) System.AggregateException 发生一个或多个错误。
