storm DRPC 예

1, DRPC 원리
클라이언트는 DRPC 서버에 실행할 방법의 이름과 이 방법의 매개 변수를 보냅니다.이 함수를 실현한 topology는 DRPCSpout를 사용하여 DRPC 서버에서 함수 호출 흐름을 수신합니다.각 함수 호출은 DRPC 서버에 고유한 id로 표시됩니다.이 topology는 결과를 계산합니다. topology의 마지막 Return Results라는 bolt는 DRPC 서버에 연결되고 이 호출된 결과를 DRPC 서버에 보냅니다.DRPC 서버는 기다리는 클라이언트와 일치하는 유일한 id로 이 클라이언트를 깨우고 결과를 보냅니다.
2, 예

package storm.starter;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.LocalDRPC;
import backtype.storm.drpc.DRPCSpout;
import backtype.storm.drpc.ReturnResults;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;


public class ManualDRPC {
  public static class ExclamationBolt extends BaseBasicBolt {

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
      declarer.declare(new Fields("result", "return-info"));
    }

    @Override
    public void execute(Tuple tuple, BasicOutputCollector collector) {
      String arg = tuple.getString(0);
      Object retInfo = tuple.getValue(1);
      collector.emit(new Values(arg + "!!!", retInfo));
    }

  }

  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"));

  }
}

참고: DRPCSpout의 이름과 drpc.execute 지정한 실행 이름 일치