Apache Storm 1.1.0未运行，无法从客户端sessionid读取其他数据

Question

我正在Apache Storm 1.1.0中运行一个简单的Hello World应用程序。应用程序有一个随机整数喷口和一个打印元组输出的螺栓。但不知何故，我无法让它在我的windows系统上工作。

public static void runTopology() {
    //String filePath = "./src/main/resources/operations.txt";
    TopologyBuilder builder = new TopologyBuilder();
    builder.setSpout("randomNumberSpout", new RandomIntSpout());
    builder.setBolt("printingBolt", new PrintingBolt()).shuffleGrouping("randomNumberSpout");

    Config config = new Config();
    config.setDebug(true);
    LocalCluster cluster = new LocalCluster();
    try{
        cluster.submitTopology("Test", config, builder.createTopology());
    }finally{
        cluster.shutdown();
    }
}

public class PrintingBolt extends BaseBasicBolt {

    /**
     * 
     */
    private static final long serialVersionUID = 1L;

    public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) {
        System.out.println("Printing Tupple!!!!");
        System.out.println(tuple);
        System.out.println("Tupple processed " + tuple.getInteger(1));
        basicOutputCollector.emit(new Values(tuple.getInteger(1)));
    }

    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("TestOutput"));

    }
}

public class RandomIntSpout extends BaseRichSpout {

    /**
     * 
     */
    private static final long serialVersionUID = 1L;

    private Random random;
    private SpoutOutputCollector outputCollector;

    /*@Override
    public void open(Map<String,Object> map, TopologyContext topologyContext,
      SpoutOutputCollector spoutOutputCollector) {
        random = new Random();
        outputCollector = spoutOutputCollector;
    }*/

    public void nextTuple() {
        Utils.sleep(1000);
        outputCollector.emit(new Values(random.nextInt(), System.currentTimeMillis()));
    }

    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("randomInt", "timestamp"));
    }

    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
            random = new Random();
            outputCollector = collector;
    }
}

我可以提供其余的代码以及，但我不认为这将是必要的。如果需要请在评论中提到，我也会提供。

当我尝试运行应用程序时，我会得到以下错误。

10620[main]信息O.A.S.S.O.A.Z.Zookeeper-启动客户端连接，connectstring=localhost:2000/storm sessiontimeout=20000>watcher=org.apache.storm.shade.org.apache.curator.connectionstate@31b0f02 10625[main-sendthread(0:0:0:0:0:0:0:0:1:2000)]信息O.A.S.S.O.A.Z.ClientCNXN-打开到服务器的套接字连接将不尝试使用SASL（未知错误）10627[Main-SendThread(0:0:0:0:0:0:0:1:2000)]信息O.A.S.S.O.A.Z.ClientCNXN-建立到>0:0:0:0:0:0:0:0:0:0:0:0:0:0:0:1:2000的套接字连接，启动会话10627[NIOServerCXN.工厂:0.0.0.0/0.0.0.0:2000]信息O.A.S.S.O.A.Z.S.NIOServerCNXN工厂-接受的套接字连接来自/>0:0:0:0:0:0:0:1:56905 10628[NIOServerCXN工厂:0.0.0.0/0.0.0.0:2000]信息O.A.S.S.O.A.Z.S.ZOOKEEPERServer-客户端试图在/>0:0:0:0:0:0:0:0:1:1:56905 10631[Main-SendThread(0:0:0:0:0:0:0:1:2000)]建立新会话信息O.A.S.S.O.A.Z.ClientCNXN-服务器上完成会话建立>0:0:0:0:0:0:0:0:0:0:1:2000，sessionid=0x16A8E5ABD97000D，协商超时=20000 10631[SyncThread:0]信息O.A.S.S.O.A.Z.S.ZookeePerServer-建立的会话0x16A8E5ABD97000D，客户端协商超时20000/>0:0:0:0:0:0:0:0:1:56905

我不明白为什么客户端套接字是关闭的，为什么会话是关闭的？我不能让它工作。请帮忙。

Answer 1

我想你可能需要在这里多睡一觉

try{
        cluster.submitTopology("Test", config, builder.createTopology());
        //Sleep here
    }finally{
        cluster.shutdown();
    }

当前您正在提交拓扑，并立即关闭。除非您Rest一下，否则您的拓扑没有机会运行。