import backtype.storm.drpc.ReturnResults; //导入依赖的package包/类
public static void main(String[] args) {
  TopologyBuilder builder = new TopologyBuilder();
  LocalDRPC drpc = new LocalDRPC();

  DRPCSpout spout = new DRPCSpout("exclamation", drpc);
  builder.setSpout("drpc", spout);
  builder.setBolt("exclaim", new ExclamationBolt(), 3).shuffleGrouping("drpc");
  builder.setBolt("return", new ReturnResults(), 3).shuffleGrouping("exclaim");

  LocalCluster cluster = new LocalCluster();
  Config conf = new Config();
  cluster.submitTopology("exclaim", conf, builder.createTopology());

  System.out.println(drpc.execute("exclamation", "aaa"));
  System.out.println(drpc.execute("exclamation", "bbb"));

}
 

import backtype.storm.drpc.ReturnResults; //导入依赖的package包/类
public static void main(String[] args) {
    TopologyBuilder builder = new TopologyBuilder();
    LocalDRPC drpc = new LocalDRPC();
    
    DRPCSpout spout = new DRPCSpout("exclamation", drpc);
    builder.setSpout("drpc", spout);
    builder.setBolt("exclaim", new ExclamationBolt(), 3)
            .shuffleGrouping("drpc");
    builder.setBolt("return", new ReturnResults(), 3)
            .shuffleGrouping("exclaim");
    
    LocalCluster cluster = new LocalCluster();
    Config conf = new Config();
    cluster.submitTopology("exclaim", conf, builder.createTopology());
    
    System.out.println(drpc.execute("exclamation", "aaa"));
    System.out.println(drpc.execute("exclamation", "bbb"));
    
}
 

import backtype.storm.drpc.ReturnResults; //导入依赖的package包/类
public static void testDrpc() {
    TopologyBuilder builder = new TopologyBuilder();
    LocalDRPC drpc = new LocalDRPC();
    
    DRPCSpout spout = new DRPCSpout("exclamation", drpc);
    builder.setSpout("drpc", spout);
    builder.setBolt("exclaim", new ExclamationBolt(), 3).shuffleGrouping("drpc");
    builder.setBolt("return", new ReturnResults(), 3).shuffleGrouping("exclaim");
    
    LocalCluster cluster = new LocalCluster();
    Config conf = new Config();
    cluster.submitTopology("exclaim", conf, builder.createTopology());
    
    JStormUtils.sleepMs(30 * 1000);
    
    try {
        System.out.println(drpc.execute("exclamation", "aaa"));
        System.out.println(drpc.execute("exclamation", "bbb"));
    } catch (Exception e) {
        Assert.fail("Failed to test drpc");
    }
    
    drpc.shutdown();
    cluster.shutdown();
}
 

import backtype.storm.drpc.ReturnResults; //导入依赖的package包/类
@Override
public TopologyBuilder createBuilder(){
	TopologyBuilder builder = super.initBuilder();
	//设置预处理步骤
	builder.setBolt(this.prepareId, new PrepareRequest()).noneGrouping(spoutId);
	AbstractDrpcBolt lastBolt = setBolt(builder, this.prepareId);
	
	OutputFieldsGetter getter = new OutputFieldsGetter();
       lastBolt.declareOutputFields(getter);
       Map<String, StreamInfo> streams = getter.getFieldsDeclaration();
       if(streams.size()!=1) {
           throw new RuntimeException("Must declare exactly one stream from last bolt in LinearDRPCTopology");
       }
       String outputStream = streams.keySet().iterator().next();
       List<String> fields = streams.get(outputStream).get_output_fields();
       if(fields.size()!=2) {
           throw new RuntimeException("Output stream of last component in LinearDRPCTopology must contain exactly two fields. The first should be the request id, and the second should be the result.");
       }
	
       //合并步骤
	builder.setBolt(this.joinId, new JoinResult(this.prepareId))
       .fieldsGrouping(lastBolt.getBoltName(), outputStream, new Fields(fields.get(0)))
       .fieldsGrouping(this.prepareId, PrepareRequest.RETURN_STREAM, new Fields("request"));
	//return步骤
	builder.setBolt(returnId, new ReturnResults()).noneGrouping(this.joinId);
	return builder;
}