1. 建立kafka spout

public TransactionalTridentKafkaSpout kafkaSpout(String topic) {
    StormConfig stormConfig = StormConfig.getInstance();

    BrokerHosts hosts = new ZkHosts(stormConfig.getZookeeper());
    TridentKafkaConfig config = new TridentKafkaConfig(hosts, topic);
    config.scheme = new SchemeAsMultiScheme(new
 
 StringScheme());

    return new TransactionalTridentKafkaSpout(config);
}

說明：建立TridentKafkaConfig時，上例中傳遞的是zookeeper的地址。實際傳遞kafka broker地址也可以。KafkaUtils中相容了兩種配置，相關原始碼：

public static IBrokerReader makeBrokerReader(Map stormConf, KafkaConfig conf) {
    if (conf.hosts instanceof StaticHosts) {
        return
 
 new StaticBrokerReader(conf.topic, ((StaticHosts) conf.hosts).getPartitionInformation());
    } else {
        return new ZkBrokerReader(stormConf, conf.topic, (ZkHosts) conf.hosts);
    }
}

2. 建立Trideng topology

TridentTopology topology = new TridentTopology();

Stream stream1 = topology.newStream("kafkaspout"
 
, kafkaSpout(topic));
stream1.each(new Fields("str"), new APacketParser(), new Fields("a-value", "a-description")).each(new Fields("a-value", "a-description"), new ProcFunction(), new Fields("whatever"));

StormTopology stormTopology = topology.build();
StormSubmitter.submitTopology("trident", new Config(), stormTopology);

說明：從kafka中讀取資訊，submitTopology時不用給任何kafka相關的配置。