Storm中DRPC如何使用

今天就跟大家聊聊有关Storm中DRPC如何使用，可能很多人都不太了解，为了让大家更加了解，小编给大家总结了以下内容，希望大家根据这篇文章可以有所收获。

1. 修改所有节点的 storm.yaml 配置文件，设置 drpc server

[grid@hadoop6 ~]$ vim storm-0.9.4/conf/storm.yamldrpc.servers:     - "hadoop4"

2. 启动drcp服务

storm drpc &

3. 编写服务端程序

//来源于 https://github.com/nathanmarz/storm-starter/blob/master/src/jvm/storm/starter/BasicDRPCTopology.javapackage storm.starter;import backtype.storm.Config;import backtype.storm.LocalCluster;import backtype.storm.LocalDRPC;import backtype.storm.StormSubmitter;import backtype.storm.drpc.LinearDRPCTopologyBuilder;import backtype.storm.topology.BasicOutputCollector;import backtype.storm.topology.OutputFieldsDeclarer;import backtype.storm.topology.base.BaseBasicBolt;import backtype.storm.tuple.Fields;import backtype.storm.tuple.Tuple;import backtype.storm.tuple.Values;public class BasicDRPCTopology {    public static class ExclaimBolt extends BaseBasicBolt {        @Override        public void execute(Tuple tuple, BasicOutputCollector collector) {            String input = tuple.getString(1);            collector.emit(new Values(tuple.getValue(0), input + "!"));        }        @Override        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);        Config conf = new Config();        if (args == null || args.length == 0) {            LocalDRPC drpc = new LocalDRPC();            LocalCluster cluster = new LocalCluster();            cluster.submitTopology("drpc-demo", conf, builder.createLocalTopology(drpc));            for (String word : new String[]{"hello", "goodbye"}) {                System.out.println("Result for \"" + word + "\": " + drpc.execute("exclamation", word));            }            cluster.shutdown();            drpc.shutdown();        } else {            conf.setNumWorkers(3);            StormSubmitter.submitTopology(args[0], conf, builder.createRemoteTopology());        }    }}

4. 将程序打包提交到Storm集群

storm jar Storm04-1.0-SNAPSHOT.jar storm.starter.BasicDRPCTopology exclamation

5. 编写客户端程序远程调用服务

import backtype.storm.utils.DRPCClient;public class Client {    public static void main(String[] args) throws Exception {        DRPCClient client = new DRPCClient("192.168.0.106", 3772);        String[] words = {"hello", "storm", "drpc"};        for (String word : words) {            String result = client.execute("exclamation", word);            System.out.println("Result for \"" + word + "\": " + result);        }    }}

6. 程序运行结果：

看完上述内容，你们对Storm中DRPC如何使用有进一步的了解吗？如果还想了解更多知识或者相关内容，请关注行业资讯频道，感谢大家的支持。