作者: xumingming | 可以转载, 但必须以超链接形式标明文章原始出处和作者信息及版权声明
网址: http://xumingming.sinaapp.com/756/twitter-storm-drpc/

新建一个Bolat作为数据处理

/**
* ExclaimBolt.java
* 版权所有(C) 2013
* 创建:cuiran 2013-01-15 17:26:42
*/
package com.stormdemo.drpc;
import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import com.stormdemo.demo.DemoTopology;
import junit.framework.TestCase;
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;
/**
* TODO
* @author cuiran
* @version TODO
*/
public class ExclaimBolt   implements IBasicBolt {
private static Log log = LogFactory.getLog(ExclaimBolt.class.getName());
/* (non-Javadoc)
* @see backtype.storm.topology.IBasicBolt#cleanup()
*/
@Override
public void cleanup() {
// TODO Auto-generated method stub
}
/* (non-Javadoc)
* @see backtype.storm.topology.IBasicBolt#execute(backtype.storm.tuple.Tuple, backtype.storm.topology.BasicOutputCollector)
*/
@Override
public void execute(Tuple tuple, BasicOutputCollector collector) {
// TODO Auto-generated method stub
log.debug("处理数据");
String input = tuple.getString(1);
log.debug("接收到的数据为:"+input);
collector.emit(new Values(tuple.getValue(0), input + "!"));
}
/* (non-Javadoc)
* @see backtype.storm.topology.IBasicBolt#prepare(java.util.Map, backtype.storm.task.TopologyContext)
*/
@Override
public void prepare(Map stormConf, TopologyContext context) {
// TODO Auto-generated method stub
}
/* (non-Javadoc)
* @see backtype.storm.topology.IComponent#declareOutputFields(backtype.storm.topology.OutputFieldsDeclarer)
*/
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
// TODO Auto-generated method stub
declarer.declare(new Fields("id", "result"));
}
/* (non-Javadoc)
* @see backtype.storm.topology.IComponent#getComponentConfiguration()
*/
@Override
public Map<String, Object> getComponentConfiguration() {
// TODO Auto-generated method stub
return null;
}
}

然后写一个测试类

/**
* DRPCTest.java
* 版权所有(C) 2013
* 创建:cuiran 2013-01-15 17:25:37
*/
package com.stormdemo.drpc;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.LocalDRPC;
import backtype.storm.drpc.LinearDRPCTopologyBuilder;
import backtype.storm.topology.TopologyBuilder;
import junit.framework.TestCase;
/**
* TODO
* @author cuiran
* @version TODO
*/
public class DRPCTest extends TestCase {
private static Log log = LogFactory.getLog(DRPCTest.class.getName());
public void testDrpc(){
log.debug("testDrpc开始");
Config conf = new Config();
conf.setDebug(true);
conf.setNumWorkers(2);
conf.setMaxSpoutPending(1);
LocalDRPC drpc=new LocalDRPC();
LocalCluster cluster=new LocalCluster();
LinearDRPCTopologyBuilder builder  = new LinearDRPCTopologyBuilder("exclamation");
builder.addBolt(new ExclaimBolt(), 3);
cluster.submitTopology("DRPCTest", conf, builder.createLocalTopology(drpc));
log.debug("传入参数返回的结果:"+drpc.execute("exclamation", "hello"));
cluster.shutdown();
drpc.shutdown();
}
}

运行结果如图：