TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("sentences", new KafkaSpout(spoutConfig), spoutNum);
builder.setBolt("split", new SplitSentence(), 10)
    .shuffleGrouping("sentences");
builder.setBolt("count", new WordCount(), 20)
    .fieldsGrouping("split", new Fields("word"));

/**
     * 想实现可靠的 Spout，需要实现如下两点
     * 1. 在 nextTuple 函数中调用 emit 函数时需要带一个
     msgId，用来表示当前的消息
（如果消息发送失败会用 msgId 作为参数回调 fail 函数）
     * 2. 自己实现 fail 函数，进行重发
（注意，在 storm 中没有 msgId 和消息的对应关系，需要自己进行维护）
     */
public void nextTuple() {
    //设置 msgId 和 Value 一样，方便 fail 之后重发
    collector.emit
(new Values(curNum + "", round +     ""), curNum + ":" + round);
}

@Override
public void fail(Object msgId) {//消息发送失败时的回调函数
String tmp = (String)msgId;
//上面我们设置了 msgId 和消息相同，这里通过 msgId 解析出具体的消息
String[] args = tmp.split(":");

//消息进行重发
collector.emit(new Values(args[0], args[1]), msgId);
}

//BaseRichBolt 实现不可靠消息传递
public class SplitSentence extends BaseRichBolt {//不建立 anchor 树的例子
    OutputCollector _collector;

public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
_collector = collector;
}

public void execute(Tuple tuple) {
String sentence = tuple.getString(0);
for(String word: sentence.split(" ")) {
_collector.emit(new Values(word)); // 不建立 anchor 树
}
_collector.ack(tuple); //手动 ack，如果不建立 anchor 树，是否 ack 是没有区别的，这句可以进行注释
}

public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("word"));
}
}

//BaseRichBolt 实现可靠的 Bolt
public class SplitSentence extends BaseRichBolt {//建立 anchor 树以及手动 ack 的例子
OutputCollector _collector;

public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
_collector = collector;
}

public void execute(Tuple tuple) {
String sentence = tuple.getString(0);
for(String word: sentence.split(" ")) {
_collector.emit(tuple, new Values(word)); // 建立 anchor 树
}
_collector.ack(tuple); //手动 ack，如果想让 Spout 重发该 Tuple，则调用 _collector.fail(tuple);
}

public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("word"));
}
}

下面的示例会可以建立 Multi-anchoring
List<Tuple> anchors = new ArrayList<Tuple>();
anchors.add(tuple1);
anchors.add(tuple2);
_collector.emit(anchors, new Values(1, 2, 3));

//BaseBasicBolt 是吸纳可靠的消息传递
public class SplitSentence extends BaseBasicBolt {//自动建立 anchor，自动 ack
public void execute(Tuple tuple, BasicOutputCollector collector) {
String sentence = tuple.getString(0);
for(String word: sentence.split(" ")) {
collector.emit(new Values(word));
}
}

public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("word"));
}
}

OpaqueTridentKafkaSpout spout =
  new OpaqueTridentKafkaSpout(spoutConf);  
  //Opaque Spout
    //TransactionalTridentKafkaSpout spout = 
 new TransactionalTridentKafkaSpout(spoutConf);  
  //Transaction Spout

TridentTopology topology = new TridentTopology();
String spoutTxid =
Utils.kafkaSpoutGroupIdBuilder
(topologyConfig.kafkaSrcTopic, topologyConfig.topologyName);
Stream stream = topology.newStream(spoutTxid, spout)
.name("new stream")
.parallelismHint(1);

// kafka config
KafkaProducerConfig kafkaProducerConfig
= new KafkaProducerConfig();
//KafkaProducerConfig 仅对 kafka
相关配置进行了封装，具体可以参考
TridentKafkaStateFactory2
(Map<String, String> config)
Map<String, String>
kafkaConfigs =
kafkaProducerConfig.loadFromConfig(topologyConfig);
TridentToKafkaMapper
tridentToKafkaMapper
= new TridentToKafkaMapper();
//TridentToKafkaMapper
继承自 TridentTupleToKafkaMapper<String, String>，
实现 getMessageFromTuple 接口，该接口中返回 tridentTuple.getString(0);

String dstTopic = "test__topic_for_all";

TridentKafkaStateFactory2 stateFactory
= new TridentKafkaStateFactory2(kafkaConfigs);
stateFactory.withTridentTupleToKafkaMapper(tridentToKafkaMapper);
stateFactory.withKafkaTopicSelector(new DefaultTopicSelector(dstTopic));

stream.each(new Fields("bytes"),
new AddMarkFunction(), new Fields("word"))
//从spout 出来数据是一个 bytes 类型的数据
，第二个是参数是自己的处理函数，第三个参数是处理函数的输出字段
.name("write2kafka")
.partitionPersist(stateFactory
//将数据写入到 Kafka 中，可以保证写入到 Kafka 的数据是 exactly once 的
, new Fields("word")
, new TridentKafkaUpdater())
.parallelismHint(1);