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我被远程调用了