WordCount是很多分布式计算中，最常用的例子，例如Hadoop、Storm，Iveely Computing也不例外。明白了WordCount在Iveely Computing上的运行原理，就很容易写出新的分布式程序。上一篇中已经知道了如何部署Iveely Computing以及提交任务，现在我们将深入WordCount的代码。
一、代码结构
图3-1
从图3-1中，可以看出，类WordCount中，有两个子类WordInput、WordOutput，以及一个主方法，WordCount.java即是一个Topology，里面至少包涵一个Input和Output（缺一不可，否则没有意义），以及main函数，main函数依然是Topology的入口函数。
现在问题来了，Input和Output到底是什么关系？还有Topology？
每一个Topology就是一个完整的任务链，可以包含多个Input，多个Output，Input的数据只能传递给一个或多个Output，Output只能将数据传递给一个或多个Output，从而形成一个完整的拓扑结构。
二、Input 深入
Input是数据的产生源，通过类WordInput看下是如何产生数据，并传递给Output的。
public static class WordInput extends IInput {
/**
* Output data to collector.
*/
private StreamChannel _channel;
/**
* All sample words.
*/
private final String[] _words = new String[] { "welcome", "iveely", "computing", "0.9.0", "build", "by",
"liufanping", "thanks", "github.com" };
private int _index;
@Override
public void start(HashMap<String, Object> conf, StreamChannel channel) {
// Here,must be initialize channel.
_channel = channel;
_index = _words.length - 1;
}
@Override
public void declareOutputFields(FieldsDeclarer declarer) {
declarer.declare(new String[] { "word" }, new Integer[] { 0 });
}
@Override
public void nextTuple() {
if (_index < 0) {
_channel.emitEnd();
} else {
for (int i = 0; i < 100; i++) {
_channel.emit(_words[_index]);
}
_index--;
}
}
@Override
public void end() {
System.out.println(getName() + " finished.");
}
@Override
public void toOutput() {
_channel.addOutputTo(new WordOutput());
}
}
函数讲解：
| start函数 | 在执行此Input之前提前调用的函数,用户初始化等相关工作,类似于构造函数，对有数据输出的时候，一定要初始化channel。 |
| declareOutputFields函数 | 用于声明输出的数据信息。 |
| nextTuple函数 | 此函数将会被频繁调用，用于输出数据，利用channel.emit提交数据到output。 |
| end函数 | 是在Input执行完毕之后，会执行的代码，类似于析构函数。 |
| toOutput函数 | 是指定Input的数据输出到的Output。 |
上面代码中，必须注意的几个问题：
2.1 WordInput必须继承IInput。
2.2 Input中，必须在start中初始化channel，因为input一定会产生数据。
2.3 Input中，toOutput函数中，必须指定数据流向。
三、Output深入
Output是数据的处理单元，也可以是新数据的产生单元。
public static class WordOutput extends IOutput {
private TreeMap<String, Integer> _map;
@Override
public void start(HashMap<String, Object> conf, StreamChannel channel) {
_map = new TreeMap<>();
}
@Override
public void declareOutputFields(FieldsDeclarer declarer) {
declarer.declare(new String[] { "word", "totalCount" }, null);
}
@Override
public void execute(Tuple tuple) {
String word = tuple.get(0).toString();
if (_map.containsKey(word)) {
int currentCount = _map.get(word);
_map.put(word, currentCount + 1);
} else {
_map.put(word, 1);
}
}
@Override
public void end() {
// Output map to database or print.
Iterator<String> it = _map.keySet().iterator();
while (it.hasNext()) {
String key = it.next();
int value = _map.get(key);
System.out.println(getName() + ":" + key + "," + value);
}
}
@Override
public void toOutput() {
}
}
与Input相比，output中没有nextTuple函数，而是取而代之的execute函数。nextTuple是产生数据，execute是处理数据。如果execute处理完毕之后的数据也需要提交到新的output中去，则需要在execute中利用channel.emit方法提交数据，此刻toOutput中也需要指定数据流向。
此处也需要注意几个问题：
3.1 如果output需要继续传递数据，则需要在start中初始化channel。
3.2 如果当前output接受的数据源来自不同的input，且数据格式不统一，则需要自行判断数据格式，例如传递数组中，第一个用int标识是什么样的数据格式。
四、main函数
main函数，依然是Topology的执行入口，不同的是，它有两种执行方式，一个是本地模式，一个是远程执行模式。本地模式是用于调试用。
public static void main(String[] args) {
TopologyBuilder builder = new TopologyBuilder(true, WordCount.class.getName(), "WordCount");
builder.setInput(new WordInput(), 1);
builder.setOutput(new WordOutput(), 4);
builder.setSlave(2);
TopologySubmitter.submit(builder, args);
}
main函数中，主要做的工作。
4.1 新建TopologyBuilder对象，并在构造函数的第一个参数指定当前是本地模式(true)还是远程模式(false)，第二个参数，指定执行的类名，第三个参数，当前Topology的名称。
4.2 设定input和output。并指定运行的数量比（线程）。
4.3 指定在多少个节点上运行（进程）。
4.4 利用TopologySubmitter提交任务即可。
4.5 注意：在生成jar提交到服务器上运行时，一定要将TopologyBuilder的第一个参数改为远程模式(false)。