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