曲阜网站设计,公众号平台网页版登录入口,视频教学网站开发,dw友情链接怎么设置关于storm的基础,参照我这篇文章:流式计算storm 关于并发和并行,参照我这篇文章:并发和并行 关于storm的并行度解释,参照我这篇文章:storm的并行度解释 关于storm的流分组策略,参照我这篇文章:storm的流分组策略 关于storm的消息可靠机制,参照我这篇文章:storm的消息可靠机…关于storm的基础,参照我这篇文章:流式计算storm 关于并发和并行,参照我这篇文章:并发和并行 关于storm的并行度解释,参照我这篇文章:storm的并行度解释 关于storm的流分组策略,参照我这篇文章:storm的流分组策略 关于storm的消息可靠机制,参照我这篇文章:storm的消息可靠机制
storm的消息可靠机制可以确保spout发出的每条tuple消息都会被完整的处理; 主要是由spout和bolt共同完成的. 本文主要讨论storm的消息可靠机制的原理和使用
storm的可靠机制,是storm的一大亮点,那么他是如何实现的呢?
先看效果:1.spout每发一条消息,就新建一个唯一的msgId(比如UUID),然后将这条消息和这个唯一id存在map中;2.每个bolt在处理tuple后,emit的时候带上tulpe,成功,就调用ack方法,代表成功,失败就调用fail方法,代表失败;这样编写代码后,你会发现,失败的消息spout会重新发送,效果就出来了
实现原理:原理很简单,使用了异或的知识点.我们知道,任意两个相同的数字,异或的结果都是0.例如:1^10现在请跟着我的思路想:1.首先想象有个服务,叫ack,他的主要作用就是判断每条tuple信息是否都成功处理2.每个spout发送和接收成功,都要给ack发送一个数字,最后由ack计算,判断整条链路是否成功处理3.spout作为发送方,假设他要给3个bolt发送消息,分别是bolt1,bolt2,bolt3;4.假设这3个bolt最后都发给bolt4;5.假设本次要处理的消息叫做root_id;6.开始发送了;7.spout给bolt1发送消息root_id,18.spout给bolt2发送消息root_id,29.spout给bolt3发送消息root_id,310.发送完spout再给ack发送1^2^311.bolt1收到root_id,1,处理成功再给bolt4发送root_id,4;12.发送完bolt1再给ack发送1^4,处理不成功就不发送了;13.bolt2收到root_id,2,处理成功再给bolt4发送root_id,5;14.发送完bolt2再给ack发送2^5,处理不成功就不发送了;15.bolt3收到root_id,3,处理成功再给bolt4发送root_id,6;16.发送完bolt3再给ack发送3^6,处理不成功就不发送了;17.bolt4收到前3个bolt的消息,root_id,4,root_id,5,root_id,6,处理成功后分别给ack发送4,5,6,处理不成功就不发送了;18.我们站在ack的角度来看,对于root_id这条消息来说,如果所有spout和bolt都成功,那么应该会收到:1^2^3,1^4,2^5,3^6,4,5,6;19.将所有收到的数字异或操作,即:1^2^3^1^4^2^5^3^6^4^5^6,由于相同数字异或结果为0,即上面的式子的结果就是0,任意少收到哪个值,最终的结果都不会为0;20.如果ack最终计算的结果是0,那么就代表这个消息root_id处理成功了21.如果ack最终计算结果不为0,那么就代表这个消息root_id处理失败了
如何使用
举个项目中的例子: spout中: 这个类 extends BaseRichSpoutprivate OutputCollector collector;private ConcurrentHashMapUUID, Values pending;Overridepublic void open(Map map, TopologyContext topologyContext, SpoutOutputCollector collector) {this.collector collector;this.pending new ConcurrentHashMap();}Overridepublic void nextTuple() {//具体业务...Values value new Values(要传的业务数据);UUID msgId UUID.randomUUID();this.pending.put(msgId, value);this.collector.emit(value, msgId);}Overridepublic void ack(Object msgId) {//收到成功消息,就删除这条msgIdthis.pending.remove(msgId);}Overridepublic void fail(Object msgId) {//收到失败消息就重新发送一遍//一般成熟的做法是会再记录个失败次数,不会一直失败重发的this.collector.emit(this.pending.get(msgId), msgId);}
bolt中: 这个类 extends BaseRichBoltprivate OutputCollector collector;Overridepublic void prepare(Map map, TopologyContext topologyContext, OutputCollector collector) {this.collectorcollector;}Overridepublic void execute(Tuple tuple) {try {//具体业务...//注意,这里发送的时候,一定要带上tuplethis.collector.emit(tuple,new Values(业务数据));collector.ack(tuple);} catch (Exception e) {collector.fail(tuple);e.printStackTrace();}}
