我正在尝试运行带有保证消息处理功能的WordCount示例。
有一个喷口
WSpout-发出带有msgID的随机句子。
和两个螺栓
SplitSentence-将单词拆分为单词并通过锚点发出
WordCount-打印字数。
我想用下面的代码实现的是，当所有单词计数为一个句子时都会完成。必须确认对应于该句子的喷口。
我仅在最后一个螺栓WordCount上使用_collector.ack（tuple）进行确认。我看到奇怪的是，尽管在WordCount.execute（）中调用了ack（），但并未调用相应的WSpout.ack（）。默认超时后总是失败。
我真的不明白代码有什么问题。请帮助我理解问题。
任何帮助表示赞赏。
以下是完整的代码。
public class TestTopology {
public static class WSpout implements IRichSpout {
SpoutOutputCollector _collector;
Integer msgID = 0;
@Override
public void nextTuple() {
Random _rand = new Random();
String[] sentences = new String[] { "There two things benefit",
" from Storms reliability capabilities",
"Specifying a link in the",
" tuple tree is " + "called anchoring",
" Anchoring is done at ",
"the same time you emit a " + "new tuple" };
String message = sentences[_rand.nextInt(sentences.length)];
_collector.emit(new Values(message), msgID);
System.out.println(msgID + " " + message);
msgID++;
}
@Override
public void open(Map conf, TopologyContext context,
SpoutOutputCollector collector) {
System.out.println("open");
_collector = collector;
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("LINE"));
}
@Override
public void ack(Object msgID) {
System.out.println("ack ------------------- " + msgID);
}
@Override
public void fail(Object msgID) {
System.out.println("fail ----------------- " + msgID);
}
@Override
public void activate() {
// TODO Auto-generated method stub
}
@Override
public void close() {
}
@Override
public void deactivate() {
// TODO Auto-generated method stub
}
@Override
public Map<String, Object> getComponentConfiguration() {
// TODO Auto-generated method stub
return null;
}
}
public static class SplitSentence extends BaseRichBolt {
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(" ")) {
System.out.println(word);
_collector.emit(tuple, new Values(word));
}
//_collector.ack(tuple);
}
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("word"));
}
}
public static class WordCount extends BaseBasicBolt {
Map<String, Integer> counts = new HashMap<String, Integer>();
@Override
public void execute(Tuple tuple, BasicOutputCollector collector) {
System.out.println("WordCount MSGID : " + tuple.getMessageId());
String word = tuple.getString(0);
Integer count = counts.get(word);
if (count == null)
count = 0;
count++;
System.out.println(word + " ===> " + count);
counts.put(word, count);
collector.emit(new Values(word, count));
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("word", "count"));
}
}
public static void main(String[] args) throws Exception {
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("spout", new WSpout(), 2);
builder.setBolt("split", new SplitSentence(), 2).shuffleGrouping(
"spout");
builder.setBolt("count", new WordCount(), 2).fieldsGrouping("split",
new Fields("word"));
Config conf = new Config();
conf.setDebug(true);
if (args != null && args.length > 0) {
conf.setNumWorkers(1);
StormSubmitter.submitTopology(args[0], conf,
builder.createTopology());
} else {
conf.setMaxTaskParallelism(3);
LocalCluster cluster = new LocalCluster();
cluster.submitTopology("word-count", conf, builder.createTopology());
Thread.sleep(10000);
cluster.shutdown();
}
}
}
最佳答案
WordCount扩展了BaseBasicBolt，可确保元组在该螺栓中被自动确认，就像您在注释中所述。但是，SplitSentence扩展了BaseRichBolt，它要求您手动确认元组。您没有接受，所以元组超时。
关于java - 具有保证消息处理功能的WordCount，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/23114487/