Storm的ACK机制与编码实例

2024-04-26 02:18
文章标签 实例 编码 机制 ack storm

本文主要是介绍Storm的ACK机制与编码实例，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

Storm为了保证每条数据成功被处理,实现至少一次语义，通过Storm的ACK机制可以对spout产生的每一个tuple进行跟踪;

tuple处理成功是指这个Tuple以及这个Tuple产生的所有子Tuple都被成功处理, 由每一个处理bolt通过OutputCollector的方法ack(tuple)来告知storm当前bolt处理成功，最终调用spout的ack方法；

处理失败是指这个Tuple或这个Tuple产生的所有Tuple中的任意一个tuple处理失败或者超时(超时时间由Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS指定), 处理失败bolt调用OutputCollector的方法fail(tuple),来告知storm当前bolt处理失败，最终调用spout的fail方法重新发送失败的tuple,失败时storm不会自动重发失败的tuple，需要我们在spout中重新获取发送失败数据，手动重新发送一次。

Ack原理

Storm中有个特殊的task名叫acker，他们负责跟踪spout发出的每一个Tuple的Tuple树（因为一个tuple通过spout发出了，经过每一个bolt处理后，会生成一个新的tuple发送出去）。当acker（框架自启动的task）发现一个Tuple树已经处理完成了，它会发送一个消息给产生这个Tuple的那个task。对任意大的一个Tuple树，storm只需要恒定的20字节就可以进行跟踪。acker对于每个spout-tuple保存一个ack-val的校验值，它的初始值是0，然后每发射一个Tuple或Ack一个Tuple时，这个Tuple的id就要跟这个校验值异或一下，
并且把得到的值更新为ack-val的新值。那么假设每个发射出去的Tuple都被ack了，那么最后ack-val的值就一定是0。Acker就根据ack-val是否为0来判断是否完全处理，如果为0则认为已完全处理。

例如下图是一个简单的Topology：



开启Act机制

1.spout发射tuple的时候指定messageId
2.spout对发射的tuple进行缓存,否则spout无法获取发送失败的数据进行重发,
(这里到底系统里有没有缓存没有成功处理的tuple,比如接口conf.setMaxSpoutPending()是否只缓存了条数还是原始数据还要去查证一下)
3.spout要重写BaseRichSpout的fail和ack方法,spout根据messageId对于成功处理的tuple从缓存队列中删除，对于失败的tuple选择重发或做其它处理；
4.如果使用BasicBolt，BasicOutputCollector在emit新的tuple时自动与源tuple锚定，execute方法结束时源tuple会被自动ack或fail;
使用RichBolt在emit数据的时需显示指定该数据的源tuple，以保持tracker链路，即collector.emit(oldTuple, newTuple);
并且需要在execute执行成功后调用OutputCollector.ack(tuple), 当失败处理时，执行OutputCollector.fail(tuple);
5.设置acker数大于0，conf.setNumAckers(>0);

关闭Ack机制

1.在Tuple层面去掉可靠性。在发射Tuple的时候不指定MessageID来达到不不跟踪这个Tuple的目的
2.如果对于一个Tuple树里面的某一部分到底成不成功不是很关心，那么可以在Bolt发射这些Tuple的时候不锚定它们。
这样这些Tuple就不在Tuple树里面，也就不会被跟踪了。
3.把Config.TOPOLOGY_ACKERS设置成0。在这种情况下，Storm会在Spout发射一个Tuple之后马上调用Spout的ack方法，
也就是说这个Tuple树不会被跟踪。

例子程序：

  1. public class RandomSentenceSpout extends BaseRichSpout {
  2.     private SpoutOutputCollector _collector;
  3.     private Random _rand;
  4.     private ConcurrentHashMap<UUID, Values> _pending;
  5.  
  6.     public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
  7. _collector = spoutOutputCollector;
  8. _rand = new Random();
  9. _pending = new ConcurrentHashMap<UUID, Values>();
  10. }
  11.  
  12.     public void nextTuple() {
  13. Utils.sleep(1000);
  14. String[] sentences = new String[] {
  15.                 "I write php",
  16.                 "I learning java",
  17.                 "I want to learn swool and tfs"
  18. };
  19. String sentence = sentences[_rand.nextInt(sentences.length)];
  20. Values v = new Values(sentence);
  21. UUID msgId = UUID.randomUUID();
  22.         this._pending.put(msgId, v);//spout对发射的tuple进行缓存
  23. _collector.emit(v, msgId);//发射tuple时，添加msgId
  24.  
  25. }
  26.  
  27.     public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
  28. outputFieldsDeclarer.declare(new Fields("world"));
  29. }
  30.  
  31.     public void ack(Object msgId) {
  32.         this._pending.remove(msgId);//对于成功处理的tuple从缓存队列中删除
  33.  
  34. }
  35.  
  36.     public void fail(Object msgId) {
  37.         this._collector.emit(this._pending.get(msgId), msgId);//当消息处理失败了，重新发射，当然也可以做其他的逻辑处理
  38.  
  39. }
  40. }
  41.  
  42. public class SplitSentence extends BaseRichBolt {
  43. OutputCollector _collector;
  44.  
  45.     public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
  46. _collector = outputCollector;
  47. }
  48.  
  49.     public void execute(Tuple tuple) {
  50. String sentence = tuple.getString(0);
  51.         for (String word : sentence.split(" "))
  52. _collector.emit(tuple, new Values(word));//发射tuple时进行锚定
  53.  
  54. _collector.ack(tuple);//对处理完的tuple进行确认
  55.  
  56. }
  57.  
  58.     public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
  59. outputFieldsDeclarer.declare(new Fields("word"));
  60. }
  61. }

场景分析

1. Bolt挂掉了，一个tuple没有被ack,storm的超时机制在超时之后会把这个tuple标记为失败，从而可以重新处理。
2. Acker挂掉了： 这种情况下由这个acker所跟踪的所有spout tuple都会超时，也就会被重新处理。
3. Spout挂掉了： 在这种情况下给spout发送消息的消息源负责重新发送这些消息。
以上机制保证storm的高度容错性

另外Ack机制还常用于限流作用： 为了避免spout发送数据太快，而bolt处理太慢，常常设置pending数，当spout有等于或超过pending数的tuple没有收到ack或fail响应时，跳过执行nextTuple， 从而限制spout发送数据。
通过conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, pending);设置spout pend数。

本文参考：
http://www.cnblogs.com/intsmaze/p/5918087.html
http://itindex.net/detail/55385-storm-trident-%E5%AD%A6%E4%B9%A0
http://www.tuicool.com/articles/vErmIb


这篇关于Storm的ACK机制与编码实例的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/936443。

相关文章

MySQL的配置文件详解及实例代码

MySQL的配置文件详解及实例代码

《MySQL的配置文件详解及实例代码》MySQL的配置文件是服务器运行的重要组成部分,用于设置服务器操作的各种参数,下面：本文主要介绍MySQL配置文件的相关资料,文中通过代码介绍的非常详细,需要... 目录前言一、配置文件结构1.[mysqld]2.[client]3.[mysql]4.[mysqldum
阅读更多...
Java 中编码与解码的具体实现方法

Java 中编码与解码的具体实现方法

《Java中编码与解码的具体实现方法》在Java中,字符编码与解码是处理数据的重要组成部分,正确的编码和解码可以确保字符数据在存储、传输、读取时不会出现乱码,本文将详细介绍Java中字符编码与解码的... 目录Java 中编码与解码的实现详解1. 什么是字符编码与解码？1.1 字符编码（Encoding）1
阅读更多...
Java Stream流以及常用方法操作实例

Java Stream流以及常用方法操作实例

《JavaStream流以及常用方法操作实例》Stream是对Java中集合的一种增强方式,使用它可以将集合的处理过程变得更加简洁、高效和易读,：本文主要介绍JavaStream流以及常用方法... 目录一、Stream流是什么？二、stream的操作2.1、stream流创建2.2、stream的使用2.
阅读更多...
深入理解go中interface机制

深入理解go中interface机制

《深入理解go中interface机制》本文主要介绍了深入理解go中interface机制,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学... 目录前言interface使用类型判断总结前言go的interface是一组method的集合，不
阅读更多...
springboot项目中集成shiro+jwt完整实例代码

springboot项目中集成shiro+jwt完整实例代码

《springboot项目中集成shiro+jwt完整实例代码》本文详细介绍如何在项目中集成Shiro和JWT,实现用户登录校验、token携带及接口权限管理,涉及自定义Realm、ModularRe... 目录简介目的需要的jar集成过程1.配置shiro2.创建自定义Realm2.1 LoginReal
阅读更多...
Python跨文件实例化、跨文件调用及导入库示例代码

Python跨文件实例化、跨文件调用及导入库示例代码

《Python跨文件实例化、跨文件调用及导入库示例代码》在Python开发过程中,经常会遇到需要在一个工程中调用另一个工程的Python文件的情况,：本文主要介绍Python跨文件实例化、跨文件调... 目录1. 核心对比表格（完整汇总）1.1 自定义模块跨文件调用汇总表1.2 第三方库使用汇总表1.3 导
阅读更多...
C# async await 异步编程实现机制详解

C# async await 异步编程实现机制详解

《C#asyncawait异步编程实现机制详解》async/await是C#5.0引入的语法糖,它基于**状态机（StateMachine）**模式实现,将异步方法转换为编译器生成的状态机类,本... 目录一、async/await 异步编程实现机制1.1 核心概念1.2 编译器转换过程1.3 关键组件解析
阅读更多...
Redis客户端连接机制的实现方案

Redis客户端连接机制的实现方案

《Redis客户端连接机制的实现方案》本文主要介绍了Redis客户端连接机制的实现方案,包括事件驱动模型、非阻塞I/O处理、连接池应用及配置优化,具有一定的参考价值,感兴趣的可以了解一下... 目录1. Redis连接模型概述2. 连接建立过程详解2.1 连php接初始化流程2.2 关键配置参数3. 最大连
阅读更多...
Spring Security 单点登录与自动登录机制的实现原理

Spring Security 单点登录与自动登录机制的实现原理

《SpringSecurity单点登录与自动登录机制的实现原理》本文探讨SpringSecurity实现单点登录（SSO）与自动登录机制,涵盖JWT跨系统认证、RememberMe持久化Token... 目录一、核心概念解析1.1 单点登录（SSO）1.2 自动登录（Remember Me）二、代码分析三、
阅读更多...
Go语言并发之通知退出机制的实现

Go语言并发之通知退出机制的实现

《Go语言并发之通知退出机制的实现》本文主要介绍了Go语言并发之通知退出机制的实现,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧... 目录1、通知退出机制1.1 进程/main函数退出1.2 通过channel退出1.3 通过cont
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2