保证消息处理的字计数

Question

我正在尝试运行WordCount示例以保证消息处理。

有一个喷口

1. WSpout-使用msgID发出随机句子

还有两个螺栓

>

SplitSentence-在单词中拆分句子并发出锚定

字数计数-打印字数计数。

我想用下面的代码实现的是，当一个句子中的所有单词都计算完毕时。必须确认与该句子对应的喷口。

我向收藏家表示感谢。仅在最后一次bolt WordCount时确认（元组）。奇怪的是ack（）的inspite在WordCount中被调用。execute（），对应的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();
    }
}
}

Answer 1

WordCount扩展了BaseBasicBolt，确保元组在该螺栓中自动确认，如您在注释中所述。但是，SplitSession扩展了BaseRichBolt，它要求您手动确认元组。你没有应答，所以元组暂停。