Twitter Storm: DRPC学习

问题导读：
1.Storm为什么引入DRP？
2.是否能根据实例新建DRPC实例？






Storm里面引入DRPC主要是利用storm的实时计算能力来并行化CPU intensive的计算。DRPC的storm topology以函数的参数流作为输入，而把这些函数调用的返回值作为topology的输出流。

DRPC其实不能算是storm本身的一个特性， 它是通过组合storm的原语spout，bolt， topology而成的一种模式(pattern)。本来应该把DRPC单独打成一个包的， 但是DRPC实在是太有用了，所以我们我们把它和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();
        }
}
复制代码

运行结果如图：

版主你好，你看下我这样理解DRPC程序对吗
1在创建new LinearDRPCTopologyBuilder("exclamation")对象，不用去实现exclamation方法，它只是作为一个DRPC方法名而已，但是我看别人要实现这个DRPC函数，也就是参数名字的函数


2.声明的第一个bolt会接收一个两维tuple指的是

drpc.execute("exclamation", "hello")
复制代码

这段代码

还是

public void declareOutputFields(OutputFieldsDeclarer declarer) {
                declarer.declare(new Fields("id", "result")); 
        }
复制代码

这段代码

3.返回最后的结果通过这段代码？

collector.emit(new Values(tuple.getValue(0), input + "!"));
复制代码

假设我有很多的bolt，那么我在topology中如下设定中间的bolt

builder.addBolt(new myBolt(), 2)
        .fieldsGrouping(new Fields("id")); //这个id和bolt定义的id一致
复制代码

到最后一个bolt时候才使用你这个例子中用的bolt形式吗？传递方法名和参数，方法名必须和
new LinearDRPCTopologyBuilder("exclamation")这个参数名字一致

希望版主能回复我

Joker 发表于 2014-11-27 21:39
版主你好，你看下我这样理解DRPC程序对吗
1在创建new LinearDRPCTopologyBuilder("exclamation")对象，不 ...

declareOutputFields 方法，告诉Storm 集群Spout 发送了哪些字段

public void declareOutputFields(OutputFieldsDeclarer declarer) {
                declarer.declare(new Fields("id", "result"));
        }


exclamation 这只是个参数，并非函数

第三个确实是返回，后面多了叹号（！）

来源http://www.aboutyun.com/thread-10316-1-1.html

楼主不要拘泥于语句
这本书你可以看看
相信对有帮助：
链接：http://pan.baidu.com/s/1hq6Aci4 密码：y21h

版主，我在集群放部署topology。每次调用一次topology跑到一半就报错。求指导 work里报的日志
2015-10-19T20:55:54.455+0800 o.a.z.ClientCnxn [INFO] Unable to read additional data from server sessionid 0x0, likely server has closed socket, closing socket connection and attempting reconnect
2015-10-19T20:55:54.556+0800 o.a.z.ClientCnxn [INFO] Opening socket connection to server 2.2.2.12/2.2.2.12:2181. Will not attempt to authenticate using SASL (unknown error)
2015-10-19T20:55:54.556+0800 o.a.z.ClientCnxn [INFO] Socket connection established to 2.2.2.12/2.2.2.12:2181, initiating session
2015-10-19T20:55:54.557+0800 o.a.z.ClientCnxn [WARN] Session 0x0 for server 2.2.2.12/2.2.2.12:2181, unexpected error, closing socket connection and attempting reconnect
java.io.IOException: Connection reset by peer
        at sun.nio.ch.FileDispatcherImpl.read0(Native Method) ~[na:1.8.0_60]
        at sun.nio.ch.SocketDispatcher.read(SocketDispatcher.java:39) ~[na:1.8.0_60]
        at sun.nio.ch.IOUtil.readIntoNativeBuffer(IOUtil.java:223) ~[na:1.8.0_60]
        at sun.nio.ch.IOUtil.read(IOUtil.java:192) ~[na:1.8.0_60]
        at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:380) ~[na:1.8.0_60]
        at org.apache.zookeeper.ClientCnxnSocketNIO.doIO(ClientCnxnSocketNIO.java:68) ~[stormjar.jar:na]
        at org.apache.zookeeper.ClientCnxnSocketNIO.doTransport(ClientCnxnSocketNIO.java:366) ~[stormjar.jar:na]
        at org.apache.zookeeper.ClientCnxn$SendThread.run(ClientCnxn.java:1081) ~[stormjar.jar:na]
2015-10-19T20:55:54.560+0800 o.a.z.ClientCnxn [INFO] Opening socket connection to server 2.2.2.12/2.2.2.12:2181. Will not attempt to authenticate using SASL (unknown error)
2015-10-19T20:55:54.560+0800 o.a.z.ClientCnxn [INFO] Socket connection established to 2.2.2.12/2.2.2.12:2181, initiating session