用于测试的组件包括
Kafka从机器上读取文件的传送器，文件由1000行组成。

String sCurrentLine;
        br = new BufferedReader(new FileReader("D:\\jsonLogTest.txt"));
        while ((sCurrentLine = br.readLine()) != null) {
            //System.out.println(sCurrentLine);
            KeyedMessage<String, String> message =new KeyedMessage<String, String>(TOPIC,sCurrentLine);
            producer.send(message);
        }

storm consumer使用三个螺栓，螺栓一应接收流并将其分为两个不同的流（stream1和stream2）。bolttwo和boltthree应该订阅这些流(简单地说，我希望处理boltone parley中的元组，就像bolt2处理前500行和bollt处理最后500行一样。
拓扑学

builder.setSpout("line-reader-spout",kafkaSpout,1);
        builder.setBolt("bolt-one", new BoltOne(),1).shuffleGrouping("line-reader-spout");
        builder.setBolt("bolt-two", new BoltTwo(),1).shuffleGrouping("bolt-one","stream1");
        builder.setBolt("bolt-three", new BoltThree(),1).shuffleGrouping("bolt-one","stream2");

博尔通

collector.emit("stream1", new Values(input.getString(0)));
            collector.emit("stream2", new Values(input.getString(0)));
        x++;System.out.println("" + x);
        collector.ack(input);

public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        // TODO Auto-generated method stub
        outputFieldsDeclarer.declareStream("stream1", new Fields("field1"));
        outputFieldsDeclarer.declareStream("stream2", new Fields("field2"));
    }

螺栓2和螺栓3

public void execute(Tuple input) {
        String sentence = input.getString(0);
        System.out.println("*********B2*************");

    }

堆栈跟踪

*********B2*************

1

*********B3*************

2

*********B2*************
*********B3*************

3

*********B3*************
*********B2*************

4

*********B3*************
*********B2*************

5

*********B2*************
*********B3*************

6

*********B2*************
*********B3*************

7

*********B3*************
*********B2*************

完全混淆了分裂流和并行。举个例子会有帮助。
我目前提出的最新解决方案：

public void execute(Tuple input) {
        @SuppressWarnings("unused")
        String sentence = input.getString(0);
        if (x%2==0) {
            collector.emit("stream1", new Values(input.getString(0)));
        }
        else{
            collector.emit("stream2", new Values(input.getString(0)));
        }

        x++;
        collector.ack(input);
    }

我只是在偶数-奇数的基础上划分流，处理时间变为一半，而bolttwo处理一个元组，而boltthree处理另一个元组。