Storm的acker消息确认机制...
ack/fail消息确认机制(确保一个tuple被完全处理)
在spout中发射tuple的时候需要同时发送messageid，这样才相当于开启了消息确认机制
如果你的topology里面的tuple比较多的话， 那么把acker的数量设置多一点,效率会高一点。
通过config.setNumAckers(num)来设置一个topology里面的acker的数量，默认值是1。
注意： acker用了特殊的算法，使得对于追踪每个spout tuple的状态所需要的内存量是恒定的（20 bytes)
注意：如果一个tuple在指定的timeout(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS默认值为30秒)时间内没有被成功处理，那么这个tuple会被认为处理失败了。
下面代码中Bolt的execute中模拟消息的正常和失败.
import java.util.Map; import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import backtype.storm.utils.Utils; /**
* 数字累加求和
* 先添加storm依赖
*
* @author Administrator
*
*/
public class LocalTopologySumAcker { /**
* spout需要继承baserichspout，实现未实现的方法
* @author Administrator
*
*/
public static class MySpout extends BaseRichSpout{
private Map conf;
private TopologyContext context;
private SpoutOutputCollector collector; /**
* 初始化方法，只会执行一次
* 在这里面可以写一个初始化的代码
* Map conf：其实里面保存的是topology的一些配置信息
* TopologyContext context：topology的上下文，类似于servletcontext
* SpoutOutputCollector collector：发射器，负责向外发射数据(tuple)
*/
@Override
public void open(Map conf, TopologyContext context,
SpoutOutputCollector collector) {
this.conf = conf;
this.context = context;
this.collector = collector;
} int num = 1;
/**
* 这个方法是spout中最重要的方法，
* 这个方法会被storm框架循环调用，可以理解为这个方法是在一个while循环之内
* 每调用一次，会向外发射一条数据
*/
@Override
public void nextTuple() {
System.out.println("spout发射："+num);
//把数据封装到values中，称为一个tuple，发射出去
//messageid:和tuple需要是一一对应的,可以把messageid认为是数据的主键id,而tuple中的内容就是这个数据.
//messageid和tuple中的消息是一一对应的. 它们之间的关系是需要我们程序员来维护的.
//this.collector.emit(new Values(num++));
this.collector.emit(new Values(num++),num-1);//传递messageid(num-1)参数就表示开启了消息确认机制.
Utils.sleep(1000);
} @Override
public void ack(Object msgId) {
System.out.println("处理成功");
} @Override
public void fail(Object msgId) {
System.out.println("处理失败....."+msgId);
//TODO--可以选择把失败的数据重发,或者单独存储后期进行分析
//重发的方法...this.collector.emit(tuple);//这个tuple可以根据参数msgId来获得...
} /**
* 声明输出字段
*/
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
//给values中的数据起个名字，方便后面的bolt从这个values中取数据
//fields中定义的参数和values中传递的数值是一一对应的
declarer.declare(new Fields("num"));
} } /**
* 自定义bolt需要实现baserichbolt
* @author Administrator
*
*/
public static class MyBolt extends BaseRichBolt{
private Map stormConf;
private TopologyContext context;
private OutputCollector collector; /**
* 和spout中的open方法意义一样
*/
@Override
public void prepare(Map stormConf, TopologyContext context,
OutputCollector collector) {
this.stormConf = stormConf;
this.context = context;
this.collector = collector;
} int sum = 0;
/**
* 是bolt中最重要的方法，当spout发射一个tuple出来，execute也会被调用，需要对spout发射出来的tuple进行处理
*/
@Override
public void execute(Tuple input) {
try{
//input.getInteger(0);//也可以根据角标获取tuple中的数据
Integer value = input.getIntegerByField("num");
if(value == 3){
throw new Exception("value=3异常.....");
}
sum+=value;
System.out.println("和："+sum);
this.collector.ack(input);//这个表示确认消息处理成功,spout中的ack方法会被调用
}catch(Exception e) {
this.collector.fail(input);//这个表示确认消息处理失败,spout中的fail方法会被调用
e.printStackTrace();
}
} /**
* 声明输出字段
*/
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
//在这没必要定义了，因为execute方法中没有向外发射tuple，所以就不需要声明了。
//如果nextTuple或者execute方法中向外发射了tuple，那么declareOutputFields必须要声明，否则不需要声明
} }
/**
* 注意：在组装topology的时候，组件的id在定义的时候，名称不能以__开头。__是系统保留的
* @param args
*/
public static void main(String[] args) {
//组装topology
TopologyBuilder topologyBuilder = new TopologyBuilder();
topologyBuilder.setSpout("spout1", new MySpout());
//.shuffleGrouping("spout1"); 表示让MyBolt接收MySpout发射出来的tuple
topologyBuilder.setBolt("bolt1", new MyBolt()).shuffleGrouping("spout1"); //创建本地storm集群
LocalCluster localCluster = new LocalCluster();
Config config = new Config();
localCluster.submitTopology("sumTopology", config, topologyBuilder.createTopology());
} }
运行结果:
从结果可以看到Bolt1执行execute成功了通过ack 调用Spout1中的ack方法.... 失败了就通过fail 调用Spout1中的fail 方法 来达到对消息处理成功与否的追踪.
//=============================================================================================
上面的例子是一个Spout 和一个Bolt.....如果对应有1个Spout和2个Bolt 会是什么情况.....
改造上面的代码.....
/**
* 数字累加求和
* 先添加storm依赖
*
* @author Administrator
*
*/
public class LocalTopologySumAcker2 { /**
* spout需要继承baserichspout，实现未实现的方法
* @author Administrator
*
*/
public static class MySpout extends BaseRichSpout{
private Map conf;
private TopologyContext context;
private SpoutOutputCollector collector; /**
* 初始化方法，只会执行一次
* 在这里面可以写一个初始化的代码
* Map conf：其实里面保存的是topology的一些配置信息
* TopologyContext context：topology的上下文，类似于servletcontext
* SpoutOutputCollector collector：发射器，负责向外发射数据(tuple)
*/
@Override
public void open(Map conf, TopologyContext context,
SpoutOutputCollector collector) {
this.conf = conf;
this.context = context;
this.collector = collector;
} int num = 1;
/**
* 这个方法是spout中最重要的方法，
* 这个方法会被storm框架循环调用，可以理解为这个方法是在一个while循环之内
* 每调用一次，会向外发射一条数据
*/
@Override
public void nextTuple() {
System.out.println("spout发射："+num);
//把数据封装到values中，称为一个tuple，发射出去
//messageid：和tuple需要是一一对应的，可以把messageid认为是数据的主键id，而tuple中的内容就是这个数据
//messageid和tuple之间的关系是需要我们程序员维护的
this.collector.emit(new Values(num++),num-1);//传递messageid参数就表示开启了消息确认机制
Utils.sleep(1000);
} /**
* 声明输出字段
*/
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
//给values中的数据起个名字，方便后面的bolt从这个values中取数据
//fields中定义的参数和values中传递的数值是一一对应的
declarer.declare(new Fields("num"));
} @Override
public void ack(Object msgId) {
System.out.println("处理成功");
} @Override
public void fail(Object msgId) {
System.out.println("处理失败。。"+msgId);
//TODO--可以选择吧失败的数据重发，或者单独存储后期分析
}
} /**
* 自定义bolt需要实现baserichbolt
* @author Administrator
*
*/
public static class MyBolt1 extends BaseRichBolt{
private Map stormConf;
private TopologyContext context;
private OutputCollector collector; /**
* 和spout中的open方法意义一样
*/
@Override
public void prepare(Map stormConf, TopologyContext context,
OutputCollector collector) {
this.stormConf = stormConf;
this.context = context;
this.collector = collector;
} int sum = 0;
/**
* 是bolt中最重要的方法，当spout发射一个tuple出来，execute也会被调用，需要对spout发射出来的tuple进行处理
*/
@Override
public void execute(Tuple input) {
try {
//input.getInteger(0);//也可以根据角标获取tuple中的数据
Integer value = input.getIntegerByField("num");
this.collector.emit(new Values(value+"_1"));
//this.collector.emit(input,new Values(value+"_1"));//新的tuple是new Values(value+"_1") 老的tuple是input
this.collector.ack(input);//确认数据处理成功，spout中的ack方法会被调用
} catch (Exception e) {
this.collector.fail(input);//确认数据处理失败，spout中的fail方法会被调用
e.printStackTrace();
}
} /**
* 声明输出字段
*/
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
//在这没必要定义了，因为execute方法中没有向外发射tuple，所以就不需要声明了。
//如果nextT|uple或者execute方法中向外发射了tuple，那么declareOutputFields必须要声明，否则不需要声明
declarer.declare(new Fields("num_1"));
} } public static class MyBolt2 extends BaseRichBolt{
private Map stormConf;
private TopologyContext context;
private OutputCollector collector; /**
* 和spout中的open方法意义一样
*/
@Override
public void prepare(Map stormConf, TopologyContext context,
OutputCollector collector) {
this.stormConf = stormConf;
this.context = context;
this.collector = collector;
} int sum = 0;
/**
* 是bolt中最重要的方法，当spout发射一个tuple出来，execute也会被调用，需要对spout发射出来的tuple进行处理
*/
@Override
public void execute(Tuple input) {
try {
//input.getInteger(0);//也可以根据角标获取tuple中的数据
String value = input.getStringByField("num_1");
System.out.println(value);
this.collector.fail(input);//确认数据处理成功，spout中的ack方法会被调用
//this.collector.ack(input);//确认数据处理成功，spout中的ack方法会被调用
} catch (Exception e) {
//this.collector.fail(input);//确认数据处理失败，spout中的fail方法会被调用
e.printStackTrace();
}
} /**
* 声明输出字段
*/
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
//在这没必要定义了，因为execute方法中没有向外发射tuple，所以就不需要声明了。
//如果nextT|uple或者execute方法中向外发射了tuple，那么declareOutputFields必须要声明，否则不需要声明
declarer.declare(new Fields("num_1"));
} } /**
* 注意：在组装topology的时候，组件的id在定义的时候，名称不能以__开头。__是系统保留的
* @param args
*/
public static void main(String[] args) {
//组装topology
TopologyBuilder topologyBuilder = new TopologyBuilder();
topologyBuilder.setSpout("spout1", new MySpout());
//.shuffleGrouping("spout1"); 表示让MyBolt接收MySpout发射出来的tuple
topologyBuilder.setBolt("bolt1", new MyBolt1()).shuffleGrouping("spout1");
topologyBuilder.setBolt("bolt2", new MyBolt2()).shuffleGrouping("bolt1"); //创建本地storm集群
LocalCluster localCluster = new LocalCluster();
localCluster.submitTopology("sumTopology", new Config(), topologyBuilder.createTopology());
}
}
上面的代码的大体意思是 Bolt1接收Spout1的输出,接收之后在数据后面加上"_1",然后发送给Bolt2,Bolt2接收到之后直接打印.
在spout2中的execute()方法不管成功还是失败 都调用 this.collector.fail(input); 方法....也就是Spout1发射的数据在Bolt1中处理都成功了,在Bolt2中的处理都失败了.
看Spout1中的哪个方法会被执行.....也就是Spout2中调用的ack或者是fail对tuple的处理状态结果是否有影响.
运行看结果:
可以看出都是成功的...这就说明tuple的处理状态和Bolt2中ack或者是fail是没有任何的关系的......只要Bolt1中处理tuple成功了,我们就认为是处理成功了...
如果Bolt1处理失败了就认为是处理失败了.. ...现在Bolt1中发射出去的tuple是无法追踪的.....
能不能在Bolt1发射的数据中也加上一个messageid...这个在Bolt中的 this.collector.emit(new Values(value+"_1")); emit方法中是不支持传入一个messageid的.
但是这样有一种场景是有问题的. 单词计数的例子:
这个Spout后面有两个Bolt 一个SplitBolt 一个CountBolt SplitBolt 切割成一个个的单词 然后再CountBolt中进行汇总....
按照上面在SplitBolt中切割成功了,就算处理成功了...但是有可能切割之后 在CountBolt中有一些Bolt没有收到. 这样最后其实是没有成功的...
而且SpiltBolt中处理的tuple和CountBolt中的tuple之间是有关联的. 后者是在前者之上切割出来的小tuple....
我们想达到两个Bolt都处理成功了才认为是处理成功的...如何做?
上面的代码中已经包括......这里再说明一下:
Spout1中 的 this.collector.emit(input,new Values(value+"_1")); ----> this.collector.emit(input,new Values(value+"_1"));//新的tuple是new Values(value+"_1") 老的tuple是input
在Spout2中还是不管是否异常都调用.. this.collector.fail(input); 看运行结果:
运行都失败了........
这样就达到了上面的"完全处理"的要求....
完全处理:保证一个tuple以及这个tuple衍生的所有tuple都被成功处理.
在storm里面一个tuple被完全处理的意思是： 这个tuple以及由这个tuple所衍生的所有的tuple都被成功处理。
如果把Bolt2的正常对应改为 this.collector.ack(input); 失败对应 this.collector.fail(input);就回复正常了.....
如果Spout2后面还有Spout3 同样把老的tuple在emit上带上.........