昨天做了一个小调查，说看看想看些啥。大概的分布是这样的，一个1代表一个投票。看来还是2、3比较多。
11111 希望看到"算法"回复1。
111111111111 希望看到"技术细节"回复2。
111111111 希望看到"成长和读书"分享回复3。
还好多人说想看我长啥样，嘛，在我比较正经的时候，就长下面这样。
大图预警！！！！
日常呢，就长这样。
长这样。
好了切入正题，今天开始挖一个新坑，就是实现一些基于MapReduce的一些图算法，比如Pregel啊，PageRank啊，LPA啊，SLPA啊等等，坑很大，非常大，慢慢写吧，都不会讲非常难的理论问题，以代码细节为主。。
先上一个我思维拓展的时候写得java实现的MapReduce的基础版本吧，写得不是很好，我也在慢慢完善，Go语言版本的还在写，真是惭愧感觉一直在吃老本。
今天实现的一个内容是，将一个List<Integer>进行map操作变成另外一个List，然后通过reduce进行加和。
灵感来源来自于《MapReduce: Simplified Data Processing on Large Clusters 》这篇论文，大家可以看看我之前的文章，在了解完什么是Mapreduce。然后先去看看这篇论文，启发很多。
首先我们从两个接口入手，MapFunction和ReduceFunction，这是MapReduce的两个灵魂接口，由使用者去定义，这里我定义的都是最最简单的版本，暂时并没有进行泛化的能力。
MapFunction定义了一个接口，类型为V，然后通过一个叫map的方法，输出一个类型为V的值。
public interface MapFunction<V> {V map(V target);}
ReduceFunction定义了一个接口，类型为V，然后通过一个叫reduce的方法，通过聚合两个V类型的值，输出一个类型为V的值。
public interface ReduceFunction<V> {V reduce(V A,V B);}
上面两个方法定义了MapReduce的核心内容，就是任务切分和任务聚合。有小伙伴不理解这里为什么使用泛型，因为作为一个框架来说，我是不知道使用者想使用什么样的类型进行计算的（虽然这里我知道我接下来就要用Integer进行计算了），所以必须不能指定类型，否则这个框架就永远只能用Integer类型了。
那我们的map和reduce任务要跑在哪里呢？有小伙伴说跑在分布式环境里。对没错，最终目的是跑在分布式环境里。但是在这里，咱就偷个懒，先用多线程来模拟这个过程，并且使用内存来作为消息机制。
我是i5双核的CPU，经验值下面，只有两个cpu的话，创建4个线程对于性能来说比单线程好。（毕竟线程切换存在开销，控制得不好多线程肯定是比单线程慢的，不服来辩）
<pre>
public class CPUs
{public static final int threads = 4;
private static final java.util.concurrent.ExecutorService pool = Executors.newFixedThreadPool(threads);
public static Future submit(Callable task){return pool.submit(task);}
public static void execute(Runnable task){pool.execute(task);}
public static void shutdown(){pool.shutdown();}}
</pre>
好了，MapFunction有了，CPUs也有了，接下来可以开始写提交器了。任务提交器是什么东西呢，就是把一个map任务进行切分，并且交给多个线程去异步执行，然后最终把结果汇总还给客户端的一个类。下面的类都比较大，建议在电脑端看。
这个类做了什么事呢？就是把List封装起来，然后把任务分发给多个线程去执行，使用CountDownLatch来保证所有的线程都已经完成计算，然后再把结果返回给客户端。
<pre>
public class MapSubmitter<V> {private List<V> target ;private int length;public MapSubmitter(List<V> target){this.target = target;this.length = target.size();}public List<V> map(final MapFunction<V> mapFunction){final CountDownLatch countDownLatch = new CountDownLatch(length);final List<V> result = new ArrayList<V>();for(int i = 0 ; i < length ; i++) {final V current = target.get(i);final int currentIndex = i;try {Future<V> future = CPUs.submit(new Callable<V>() {public V call() throws Exception {V result = mapFunction.map(current);//Printer.println(currentIndex); return result;}});result.add(i,future.get());countDownLatch.countDown();}catch (InterruptedException e) {e.printStackTrace();} catch (ExecutionException e) {e.printStackTrace();}}try{countDownLatch.await();} catch (InterruptedException e) {}finally {return result;}}}
</pre>
这个类又做了什么事呢？List封装起来，交给很多线程去执行，然后维护一个最终的结果类V，并为这个结果提供线程安全的保护，避免因为多线程操作同一个结果造成结果错误。
<pre>
public class ReduceSubmitter<V> {private List<V> target ;private int length;private V result ;Lock lock = new ReentrantLock();public ReduceSubmitter(List<V> target){this.target = target;this.length = target.size();this.result = target.get(0);}public V reduce(final ReduceFunction<V> reduceFunction){final CountDownLatch countDownLatch = new CountDownLatch(length);countDownLatch.countDown();for(int i = 1 ; i < length ; i ++) {final V current = target.get(i);CPUs.execute(new Runnable() {public void run() {lock.lock();V next = reduceFunction.reduce(ReduceSubmitter.this.result,current);ReduceSubmitter.this.result = next;lock.unlock();countDownLatch.countDown();}});}try{countDownLatch.await();} catch (InterruptedException e) {}finally {return this.result;}}}
</pre>
好咯，写完了就开始测试了,主要就创建一个长度为10的数组，然后进行map操作把每一个值都进行平方，然后通过reduce操作进行求和，代码比较简单就不一一细说了，有啥问题后台留言交流。
<pre>
public class TestMapReduce {public static void main(String[] args){//仅仅是为了耗时而模拟的一个好像很复杂的操作，不然太快了。final int junkTime = 1000000;//初始化一个想进行操作的数组List<Integer> integerList = new ArrayList<Integer>();for(int i = 0 ; i < 10 ; i++){integerList.add(i);}int length = integerList.size();// printer.printList(integerList); Long start = System.currentTimeMillis();//进行map操作并返回结果MapSubmitter<Integer> mapSubmitter = new MapSubmitter<Integer>(integerList);integerList = mapSubmitter.map(new MapFunction<Integer>() {public Integer map(Integer target) {Double b = 0D;for(int i = 0 ; i <junkTime;i++){b += Math.exp(i);}return target * target;}});Printer.println("mapreduce cost time:" + (System.currentTimeMillis() - start)); start = System.currentTimeMillis();
//进行reduce操作并返回结果
ReduceSubmitter<Integer> reduceSubmitter = new ReduceSubmitter<Integer>(integerList);Integer resultInteger = reduceSubmitter.reduce(new ReduceFunction<Integer>() {public Integer reduce(Integer A, Integer B) {Double b = 0D;for(int i = 0 ; i <junkTime;i++){b += Math.exp(i);}return A+B;}});Printer.println("reduce cost time:" + (System.currentTimeMillis() - start)); CPUs.shutdown();}}
</pre>
好啦，今天的MapReduce就说到这里。经过我的实验，无论多少次实验，都是比单线程快那么一丢丢的，这都要得益于那个耗时的操作，模糊了线程切换带来的时间损耗，毕竟不怎么耗时的操作来说，单线程其实是绝对比多线程快的。
细心的同学会发现，好像这个并不符合论文里面的标准吖。嗯呐是的，这个只是我心血来潮写的简单版本。问题有诸如，我们上面的map操作好像不能变成其他类型吖，怎么实现WordCount呢？以及Driver好像没有进行任务切分和分发吖？好像也没有suffle操作啊？好像整个过程也不是严格多线程的吖，怎么办呢？下一次给大家分享一个更加完整的MapReduce。
希望大家都能在自己的机器上跑成功。源码都在上面了我就不放链接了。
好了，如果有任务问题请后台留言，我会看的。如果对您有一点点的帮助或者启发的话，帮忙转发或者点个赞都是对我很大的支持喔，么么哒。
