多线程流式计算模型

现在的Count引擎使用的一个框架。非常轻量级。全部使用了JDK自带的线程池和队列做实现。使用起来也相当方便。

一、QuickStart

1.

2.三大基础类

2.1 EmitItem ： 发布和处理的KV对的封装

2.2 ModeHandler ： 执行业务逻辑的单元 已改名为Bolt

2.3 FirstModeHandler ： 读取文件/队列/存储等用于生产第一批EmitItem的单元 已改名为Spout

3.拓扑类

3.1 Topology包含了一系列的ModeHandler和一种FirstModeHandler的实现。

3.2 可以按照实际的业务需求来任意组合ModeHandler的执行顺序，但是不能动态调整，每次调整都需要重新编译

3.3 拓扑的启动可以直接main方法启动，也可以放置于容器内部署启动

二、SimpleExample

1.利用map统计单词出现的次数

public class CountModeHandler extends ModeHandler {
 
...
 
 @Override
 
 public int execute() {  //业务逻辑实现方法
 
    EmitItem item = null;
 
     int num = 0;
 
     String word = null;
 
     Map<String, Integer> map = new HashMap<String, Integer>();
 
     while ((item = getReadMessageQueue().poll()) != null) {//从前置队列中获取item
 
         try {
 
            word = (String) item.getMessage(0);//从item中得到要统计的word
 
            if (map.containsKey(word))
 
                map.put(word, map.get(word) + 1);//利用map来统计word出现的次数
 
             else
 
                map.put(word, 1);
 
            num++;
 
         } catch (Exception e) {
 
            e.printStackTrace();
 
         }
 
     }
 
     if (num > 0) {
 
         emit(0, map);//将统计结果存放的map继续发射到下一个处理单元
 
     }
 
    return num;
 
 }
 
 
...
}
2.将不同统计Mode的map结果合并成一个map

public class MergeModeHandler extends ModeHandler {
 
 
...
@Override
 
 public int execute() {
 
    EmitItem item = null;
 
     int num = 0;
 
     while ((item = getReadMessageQueue().poll()) != null) {
 
         try {
 
             Map<String, Integer> m = (Map<String, Integer>) item.getMessage(0);
 
             for (String key : m.keySet()) {
 
                 if (map.containsKey(key))
 
                    map.put(key, map.get(key) + m.get(key));
 
                else
 
                     map.put(key, m.get(key));
 
            }
 
 
 
 
         } catch (Exception e) {
 
             e.printStackTrace();
 
     }
 
     num++;
 
     }
 
    return num;
 
 }
 
 
...
}
3.模拟读入文章并将单词发射出去的FirstModeHandler

public class TestFirstModeHandler extends FirstModeHandler {
 
 
    public String[][] message = new String[][] {
 
     { "one", "apple", "a", "day", "doctor", "keeps", "away" },
 
     { "when", "a", "man", "loves", "a", "woman" },
 
     { "what", "doesn't", "kill", "you", "makes", "you", "stronger" } };
 
 
    @Override
 
     public int execute() {
 
        int index = 0;
 
        for (int i = 0; i < message.length; i++)
 
         for (int j = 0; j < message[i].length; j++) {
 
             emit(index, message[i][j]);
 
             index++;
 
         }
 
         return index;
 
     }
...
 
 
}
4.将各个单元组装成拓扑结构并运行

...
public static void main(String[] args) {
 
 TestTopology tt = new TestTopology();
 
 
 
 
 TestFirstModeHandler h1 = new TestFirstModeHandler();
 
 CountModeHandler h2 = new CountModeHandler();
 
 MergeModeHandler h3 = new MergeModeHandler();
 
 tt.prepare();//拓扑的准备阶段，可以初始化一些参数
 
 h3.prepare(tt.map);//ModeHandler的准备阶段，同样可以初始化参数
 
 tt.setFirstModeHandler(h1, 1).setModeHandler(h2, 4).setModeHandler(h3, 2); //拓扑设置各个单元的数量和顺序，先set哪个ModeHandler，哪个就先被执行。
 
 tt.start();//启动
 
 System.out.println("The last result is : " + tt.map);//输入统计单词的结果
 
 tt.shutdown();//停止
 
 }
 
 
...
5.TestTopology只简单的初始化一个map用来存放最终统计的结果就可以了

public class TestTopology extends NewTopology {
 
 
 
 
 public Map<String, Integer> map;
 
 
 
 
 @Override
 
 public void prepare() {
 
 map = new ConcurrentHashMap<String, Integer>();
 
 };
 
 
 
 
}
三、TODOLIST

· 分流/整流/订阅 三种拓扑结构的实现

· 监控的一些整合