c#扩展出MapReduce方法

MapReduce方法主体：

 1 public static IDictionary<TKey, TResult> MapReduce<TInput, TKey, TValue, TResult>(this IEnumerable<TInput> inputList,
 2             Func<TInput, KeyValueClass<TKey, TValue>> map, Func<TKey, IEnumerable<TValue>, TResult> reduce)
 3         {
 4             object locker = new object();
 5             ConcurrentDictionary<TKey, TResult> result = new ConcurrentDictionary<TKey, TResult>();
 6             //保存map出来的结果
 7             ConcurrentDictionary<TKey, ConcurrentBag<TValue>> mapDic = new ConcurrentDictionary<TKey, ConcurrentBag<TValue>>();
 8             //并行map
 9             Parallel.ForEach(inputList, (t, p, s) =>
10             {
11                 var pair = map(t);
12                 if (pair != null)
13                 {
14                     //锁住防止并发操作list造成数据缺失
15                     lock (locker)
16                     {
17                         //将匹配出来的结果加入结果集放入字典
18                         ConcurrentBag<TValue> list = null;
19                         if (mapDic.ContainsKey(pair.Key))
20                         {
21                             list = mapDic[pair.Key];
22                         }
23                         else
24                         {
25                             list = new ConcurrentBag<TValue>();
26                             mapDic[pair.Key] = list;
27                         }
28                         list.Add(pair.Value);
29                     }
30                 }
31             });
32
33             //并行reduce
34             Parallel.ForEach(mapDic.Keys, (t, p, s) =>
35             {
36                 result[t] = reduce(t, mapDic[t]);
37             });
38             return result;
39         }

KeyValueClass定义：

 1 public class KeyValueClass<K, V>
 2     {
 3         public KeyValueClass(K key, V value)
 4         {
 5             Key = key;
 6             Value = value;
 7         }
 8
 9         public KeyValueClass()
10         {
11
12         }
13
14         public K Key { get; set; }
15
16         public V Value { get; set; }
17     }

Console测试：

 1 List<TestClass> listTestClass = new List<TestClass>();
 2             listTestClass.Add(new TestClass { a = "a", g = 1 });
 3             listTestClass.Add(new TestClass { a = "b", g = 3 });
 4             listTestClass.Add(new TestClass { a = "c", g = 4 });
 5             listTestClass.Add(new TestClass { a = "d", g = 2 });
 6             listTestClass.Add(new TestClass { a = "e", g = 1 });
 7             listTestClass.Add(new TestClass { a = "f", g = 2 });
 8             listTestClass.Add(new TestClass { a = "g", g = 5 });
 9             listTestClass.Add(new TestClass { a = "h", g = 6 });
10             IDictionary<int, string> dic = listTestClass.MapReduce(t =>
11             {
12                 if (t.g < 5)
13                 {
14                     return new KeyValueClass<int, string>(t.g, t.a);
15                 }
16                 return null;
17             }, (key, values) =>
18            {
19                return string.Join(",", values);
20            });

TestClass定义：

 1 public class TestClass
 2     {
 3         public string a { get; set; }
 4         public string b { get; set; }
 5
 6         public string d { get; set; }
 7
 8         //public DateTime f { get; set; }
 9
10         public int g { get; set; }
11
12         public List<TestClass> test { get; set; }
13
14         public Dictionary<string, string> dic { get; set; }
15     }

结果：

1：a,e

2：d,f

3：b

4：c