storm 远程过程调用及client调用

Posted jzdwajue

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了storm 远程过程调用及client调用相关的知识,希望对你有一定的参考价值。

server端:

import java.util.Map;
import org.apache.thrift7.TException;
import backtype.storm.Config;
import backtype.storm.ILocalDRPC;
import backtype.storm.LocalCluster;
import backtype.storm.LocalDRPC;
import backtype.storm.StormSubmitter;
import backtype.storm.drpc.DRPCSpout;
import backtype.storm.drpc.LinearDRPCTopologyBuilder;
import backtype.storm.generated.DRPCExecutionException;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;


public class RemoteDRCPTopology {


public static class drcpBolt extends BaseRichBolt {


private Map stormConf;
private TopologyContext context;
private OutputCollector collector;


public void prepare(Map stormConf, TopologyContext context,
OutputCollector collector) {
this.stormConf = stormConf;
this.context = context;
this.collector = collector;
}


/**
* 第一个 :函数名 第二个 :參数
*/


public void execute(Tuple input) {
String value = input.getString(1);
value = "hello" + value;
this.collector.emit(new Values(input.getValue(0), value));


}


public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("name", "value"));


}
}


public static void main(String[] args) throws Exception, TException {
LinearDRPCTopologyBuilder topologyBuilder = new LinearDRPCTopologyBuilder(
"aaa");

topologyBuilder.addBolt(new drcpBolt());
Map stormConf;
StormSubmitter.submitTopology(RemoteDRCPTopology.class.getSimpleName(),
new Config(), topologyBuilder.createRemoteTopology());
}


}


client:


import backtype.storm.utils.DRPCClient;

public class ClientRemoteDRPC {
public static void main(String[] args) throws Exception{
DRPCClient drpcClient = new DRPCClient("192.168.80.20", 3772);
String res = drpcClient.execute("aaa", "哈哈哈444我被远程调用了");
System.out.println(res);
}
}


结果输出为:

hello哈哈哈444我被远程调用了

以上是关于storm 远程过程调用及client调用的主要内容,如果未能解决你的问题,请参考以下文章

JSON-RPC轻量级远程调用协议介绍及使用

Hadoop RPC远程过程调用源码解析及实例

RPC实践-Thrift

SOA五种基本架构模式及远程过程调用

HDFS(二) 底层通信原理——RPC 及 动态代理

Storm介绍及与Spark Streaming对比