Big Data Road

客戶端發送功能名稱及功能所需參數到DRPC服務器去執行。圖中的拓撲實現了此功能，它使用DRPCSpout從DRPC服務器接收功能調用流。每個功能調用通過DRPC服務器使用唯一ID標記，隨后拓撲計算結果，在拓撲的最后，一個稱之為“ReturnResults”的bolt連接到DRPC服務器，把結果交給這個功能調用（根據功能調用ID），DRPC服務器根據ID找到等待中的客戶端，為等待中的客戶端消除阻塞，并發送結果給客戶端。

public static class ExclaimBolt implements IBasicBolt { 
    public void prepare(Map conf, TopologyContext context) { 
    } 
 
    public void execute(Tuple tuple, BasicOutputCollector collector) { 
        String input = tuple.getString(1); 
        collector.emit(new Values(tuple.getValue(0), input + "!")); 
    } 
 
    public void cleanup() { 
    } 
 
    public void declareOutputFields(OutputFieldsDeclarer declarer) { 
        declarer.declare(new Fields("id", "result")); 
    } 
 
} 
 
public static void main(String[] args) throws Exception { 
    LinearDRPCTopologyBuilder builder = new LinearDRPCTopologyBuilder("exclamation"); 
    builder.addBolt(new ExclaimBolt(), 3); 
    // ... 
} 

如你所見，代碼非常少。當創建LinearDRPCTopologyBuilder時，你把這個拓撲的DRPC功能名稱告訴storm。一個DRPC服務器可以協調許多功能，功能名稱用于區別不同的功能，首先聲明的bolt將接收一個輸入的2-tuples，第一個字段是請求ID，第二個字段是請求參數。LinearDRPCTopologyBuilder認為最后的bolt會發射一個輸出流，該輸出流包含[id, result]格式的2-tuples。最后，所有拓撲中間過程產生的元組（tuple）都包含請求id作為其第一個字段。

LocalDRPC drpc = new LocalDRPC(); 
LocalCluster cluster = new LocalCluster(); 
 
cluster.submitTopology("drpc-demo", conf, builder.createLocalTopology(drpc)); 
 
System.out.println("Results for 'hello':" + drpc.execute("exclamation", "hello")); 
 
cluster.shutdown(); 
drpc.shutdown(); 

bin/storm drpc 

drpc.servers: 
  - "drpc1.foo.com" 
  - "drpc2.foo.com" 

StormSubmitter.submitTopology("exclamation-drpc", conf, builder.createRemoteTopology()); 

public static class PartialUniquer implements IRichBolt, FinishedCallback { 
    OutputCollector _collector; 
    Map<Object, Set<String>> _sets = new HashMap<Object, Set<String>>(); 
     
    public void prepare(Map conf, TopologyContext context, OutputCollector collector) { 
        _collector = collector; 
    } 
 
    public void execute(Tuple tuple) { 
        Object id = tuple.getValue(0); 
        Set<String> curr = _sets.get(id); 
        if(curr==null) { 
            curr = new HashSet<String>(); 
            _sets.put(id, curr); 
        } 
        curr.add(tuple.getString(1)); 
        _collector.ack(tuple); 
    } 
 
    public void cleanup() { 
    } 
 
    public void finishedId(Object id) { 
        Set<String> curr = _sets.remove(id); 
        int count; 
        if(curr!=null) { 
            count = curr.size(); 
        } else { 
            count = 0; 
        } 
        _collector.emit(new Values(id, count)); 
    } 
 
    public void declareOutputFields(OutputFieldsDeclarer declarer) { 
        declarer.declare(new Fields("id", "partial-count")); 
    } 
}