strom远程drpc函数接口

drpc是storm的远程函数接口，实现方式有两种

此处代码仅适用集群运行，本地调试请自行修改local启动方式

1、服务器端代码，服务器代码打jar包后在服务器端运行，请注意“统一写法”

方法1

package bhz.drcp;
import backtype.storm.Config;
import backtype.storm.StormSubmitter;
import backtype.storm.drpc.DRPCSpout;
import backtype.storm.drpc.ReturnResults;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class ManualDRPC {
    //服务器端
    public static void main(String[] args) throws  Exception   {
        TopologyBuilder builder = new TopologyBuilder();
        DRPCSpout spout = new DRPCSpout("add");
        builder.setSpout("drpc", spout);
        builder.setBolt("add", new AddBolt(),3).shuffleGrouping("drpc");
        builder.setBolt("return", new ReturnResults(),3).shuffleGrouping("add");
        Config conf = new Config();
        StormSubmitter.submitTopology("ManualDRPC", conf, builder.createTopology());
    }
    public static class AddBolt extends BaseBasicBolt{
        @Override
        public void execute(Tuple input, BasicOutputCollector collector) {
            Object returnInfo = input.getValue(1);
            String params = input.getString(0);
            String conversValue = params+"!!!!";
            collector.emit(new Values(conversValue,returnInfo));
        }
        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields("result","return-info"));//统一写法，注意
        }
        
    }
}

 

方法2

 

package bhz.drcp;

import backtype.storm.Config;
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 ManualDRPC {
	public static void main(String[] args) throws Exception {
		LinearDRPCTopologyBuilder builder = new LinearDRPCTopologyBuilder("add1");
		builder.addBolt(new AddBolt(), 1);
		Config conf = new Config();
		conf.setDebug(false);
		conf.setNumWorkers(2);
		StormSubmitter.submitTopology("HelloDRPC", conf, builder.createRemoteTopology());

	}
	public static class AddBolt extends BaseBasicBolt {
		@Override
		public void execute(Tuple input, BasicOutputCollector collector) {
			Object returnInfo = input.getValue(1);
			String params = input.getString(0);
			String conversValue = params + "!!!!";
			collector.emit(new Values(conversValue, returnInfo));
		}
		@Override
		public void declareOutputFields(OutputFieldsDeclarer declarer) {
		        declarer.declare(new Fields("id","value"));//统一写法，注意
		}

	}
}

 

 

2、客户端、可以运行在任何环境

package bhz.drcp;

import org.apache.thrift7.TException;
import backtype.storm.generated.DRPCExecutionException;
import backtype.storm.utils.DRPCClient;
public class ManualClientDRPC {
    public static void main(String[] args) throws TException, DRPCExecutionException {
            DRPCClient client = new DRPCClient( "sto1", 3772);
            String result = client.execute("add", "1,asdf2");
            System.err.println(result);
    }
}