strom DRPC服务调用例子

 

strom DRPC服务调用例子


DRPC服务端代码

import java.util.Map;
import backtype.storm.Config;
import backtype.storm.StormSubmitter;
import backtype.storm.drpc.LinearDRPCTopologyBuilder;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.IBasicBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
 
@SuppressWarnings("deprecation")
public class My_storm_drpc {
    public static class storm_drpc_Bolt implements IBasicBolt {

        private static final long serialVersionUID = 3812791870691350630L;
 
        StringBuilder Str = new StringBuilder();
 
        public void prepare(Map conf, TopologyContext context) {
        }
 
        public void execute(Tuple tuple, BasicOutputCollector collector) {
            Str.append("@");
            String input = tuple.getString(1);
            collector
                    .emit(new Values(tuple.getValue(0), input + Str.toString()));
        }
 
        public void cleanup() {
        }
 
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields("id", "result"));
        }
 
        @Override
        public Map<String, Object> getComponentConfiguration() {
            // TODO Auto-generated method stub
            return null;
        }
 
    }
 
    @SuppressWarnings("deprecation")
    public static void main(String[] args) throws Exception {
 
        LinearDRPCTopologyBuilder builder = new LinearDRPCTopologyBuilder(
                "My_storm_drpc");
        builder.addBolt(new storm_drpc_Bolt(), 3).allGrouping();
        Config conf = new Config();
        conf.setDebug(false);
        conf.setNumWorkers(4);
        StormSubmitter.submitTopology(args[0], conf,
                builder.createRemoteTopology());
    }
}

 


把代码打包到了一个mystorm.jar的包里
storm测试环境集群只有两台
mynode001 部署了nimbus、zookeeper 以及 drpc server
mynode002 部署了supervisor
storm版本是twitter storm 0.9.0.1最新版

 

把mystorm.jar copy到mynode001上，

执行以下命令：storm jar /data/storm/storm-yarn-master/lib/mystorm.jar My_storm_drpc My_storm_drpc -c nimbus.host=mynode001


DRPC客户端调用

import backtype.storm.utils.DRPCClient;
 
public class My_storm_drpc_client {
    public static void main(String[] args) throws Exception {

        DRPCClient client = null;
        client = new DRPCClient("mynode001", 3772);
        System.out.println("开始执行DRPC客户端调用");
        for (int i = 0; i < 10; i++) {
            String tt = client.execute("My_storm_drpc", "你好");
            System.out.println("tt = " + tt);
        }
    }
}

 


客户机执行以下命令：storm jar /data/storm/storm-yarn-master/lib/mystorm.jar My_storm_drpc_client -c nimbus.host=mynode001

 

 

 客户端给DRPC服务器发送要执行的方法的名字，以及这个方法的参数。实现了这个函数的topology使用DRPCSpout从DRPC服务器接收函数调用流。每个函数调用被DRPC服务器标记了一个唯一的id。 这个topology然后计算结果，在topology的最后一个叫做ReturnResults的bolt会连接到DRPC服务器，并且把这个调用的结果发送给DRPC服务器(通过那个唯一的id标识)。DRPC服务器用那个唯一id来跟等待的客户端匹配上，唤醒这个客户端并且把结果发送给它。

 

storm_drpc_Bolt类中的execute方法中参数tuple，第一个field是request-id，第二个field是这个请求的参数。LinearDRPCTopologyBuilder同时要求我们topology的最后一个bolt发射一个二维tuple： 第一个field是request-id， 第二个field是这个函数的结果。最后所有中间tuple的第一个field必须是request-id，和客户端的

request-id匹配，匹配上就传回这个客户端结果。

 

详细介绍看徐明明介绍：http://xumingming.sinaapp.com/756/twitter-storm-drpc/