protected StormTopology buildTopology() {
// ConsumerConfig.
ZkHosts zkHosts = new ZkHosts(brokerZkStr, brokerZkPath);
// storm.kafka.KafkaSpout
TridentTopology tridentTopology = new TridentTopology();
TridentKafkaConfig spoutConf = new TridentKafkaConfig(zkHosts, topic);
spoutConf.forceFromStart = true;// 和startOffsetTime,一起用,默认情况下,为false,一旦startOffsetTime被设置,就要置为true
spoutConf.startOffsetTime = -1L;// -2 从kafka头开始 -1 是从最新的开始 0 =无 从ZK开始
spoutConf.scheme = new SchemeAsMultiScheme(new StringScheme());
// OpaqueTridentKafkaSpout opaqueTridentKafkaSpout = new OpaqueTridentKafkaSpout(spoutConf);
TransactionalTridentKafkaSpout opaqueTridentKafkaSpout = new TransactionalTridentKafkaSpout(spoutConf);
Stream opTrdtSptStream = tridentTopology.newStream("opTrdtSptStream", opaqueTridentKafkaSpout);
opTrdtSptStream.shuffle().each(new Fields("str"), new Utils.FilterLogInfo()).each(new Fields("str"), new
LogFilterUtil(), new Fields("analyzeResult")).each(new Fields("analyzeResult"),new PropertyAnalyzeUtils
(),new Fields("autoItem")).each(new Fields("autoItem"), new AnsjIllegalWordsUtil
(), new Fields("ansjIllegalWordsResult")).each(new Fields("ansjIllegalWordsResult"), new
AnsjKeyWordsUtil(), new Fields("ansjKeyWordsResult")).each(new Fields("ansjKeyWordsResult"), new
AnsjPicturesUtil(), new Fields("ansjPicturesResult"));
return tridentTopology.build();
}