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)。