我正在使用 KafkaSpout。请在下面找到测试程序。
我正在使用 Storm 0.8.1。Storm 0.8.2 中有 Multischeme 类。我会用那个。我只想通过实例化 StringScheme() 类来了解早期版本是如何工作的？我在哪里可以下载早期版本的 Kafka Spout？但我怀疑这不是在 Storm 0.8.2 上工作的正确选择。？？？（使困惑）
当我在storm集群上运行代码（如下所示）时（即当我推送我的拓扑时）我得到以下错误（当Scheme部分被注释时会发生这种情况，当然我会得到编译器错误，因为该类在0.8中不存在.1):
java.lang.NoClassDefFoundError: backtype/storm/spout/MultiScheme
at storm.kafka.TestTopology.main(TestTopology.java:37)
Caused by: java.lang.ClassNotFoundException: backtype.storm.spout.MultiScheme
在下面给出的代码中，您可能会发现 spoutConfig.scheme=new StringScheme(); 部分评论。如果我不评论那条很自然的行，我会收到编译器错误，因为那里没有构造函数。此外，当我实例化 MultiScheme 时，我会收到错误，因为我在 0.8.1 中没有该类。
public class TestTopology {
public static class PrinterBolt extends BaseBasicBolt {
public void declareOutputFields(OutputFieldsDeclarer declarer) {
}
public void execute(Tuple tuple, BasicOutputCollector collector) {
System.out.println(tuple.toString());
}
}
public static void main(String [] args) throws Exception {
List<HostPort> hosts = new ArrayList<HostPort>();
hosts.add(new HostPort("127.0.0.1",9092));
LocalCluster cluster = new LocalCluster();
TopologyBuilder builder = new TopologyBuilder();
SpoutConfig spoutConfig = new SpoutConfig(new KafkaConfig.StaticHosts(hosts, 1), "test", "/zkRootStorm", "STORM-ID");
spoutConfig.zkServers=ImmutableList.of("localhost");
spoutConfig.zkPort=2181;
//spoutConfig.scheme=new StringScheme();
spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
builder.setSpout("spout",new KafkaSpout(spoutConfig));
builder.setBolt("printer", new PrinterBolt())
.shuffleGrouping("spout");
Config config = new Config();
cluster.submitTopology("kafka-test", config, builder.createTopology());
Thread.sleep(600000);
}