Storm 从入门到精通 第二十讲 Storm DRPC 简单示例_John,hao的专栏

这里简单Echo程序：

1. BasicDRPCTopology

package com.john.learn.storm.drpc.basic;

import java.util.Arrays;
import java.util.Scanner;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.LocalDRPC;
import org.apache.storm.StormSubmitter;
import org.apache.storm.drpc.DRPCSpout;
import org.apache.storm.drpc.LinearDRPCTopologyBuilder;
import org.apache.storm.drpc.ReturnResults;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.shade.org.apache.commons.exec.util.StringUtils;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import clojure.main;

public class BasicDRPCTopology {

  public static class EchoBolt extends BaseBasicBolt {

    public void execute(Tuple tuple, BasicOutputCollector collector) {

     // 你声明的第一个bolt会接收两维tuple，tuple的第一个field是request-id，第二个field是这个请求的参数

     System.out.println("EchoBolt execute input " + tuple.getValue(0) + " Input val:" + tuple.getString(1));

     collector.emit(new Values(tuple.getValue(0), "Hi " + tuple.getString(1)));
    }

    public void declareOutputFields(OutputFieldsDeclarer fieldsDeclarer) {

     fieldsDeclarer.declare(new Fields("id", "result"));
    }

    private static final long serialVersionUID = 1L;
  }

  public static void main(String[] args)
     throws AlreadyAliveException, InvalidTopologyException, AuthorizationException {

    Config config = new Config();
    config.setDebug(false);
    config.setNumWorkers(1);

    // DRPC Server Name
    LinearDRPCTopologyBuilder builder = new LinearDRPCTopologyBuilder("Echo");
    // 数据源Spout直接来此DRPCClient,不用指定Spout
    builder.addBolt(new EchoBolt(), 3);

    if (args == null || args.length == 0) {

     LocalDRPC drpc = new LocalDRPC();

     LocalCluster cluster = new LocalCluster();

     cluster.submitTopology("BasicDRPCTopology", config, builder.createLocalTopology(drpc));

     Scanner scanner = new Scanner(System.in);

     System.out.println("Please input a value:");

     if (scanner.hasNextLine()) {
       System.out.println("Response :" + drpc.execute("Echo", scanner.nextLine()));
     }

     // 便于找到Response log
     System.exit(0);

     drpc.shutdown();
     cluster.shutdown();

     return;
    }

    config.setNumWorkers(3);
    // Server Mode
    StormSubmitter.submitTopology("BasicDRPCTopology", config, builder.createRemoteTopology());

  }

}
 
 
package