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。

相关文章

JVM 的类初始化机制

JVM 的类初始化机制

前言 当你在 Java 程序中new对象时，有没有考虑过 JVM 是如何把静态的字节码（byte code）转化为运行时对象的呢，这个问题看似简单，但清楚的同学相信也不会太多，这篇文章首先介绍 JVM 类初始化的机制，然后给出几个易出错的实例来分析，帮助大家更好理解这个知识点。 JVM 将字节码转化为运行时对象分为三个阶段，分别是：loading 、Linking、initialization
阅读更多...
【机器学习】高斯过程的基本概念和应用领域以及在python中的实例

【机器学习】高斯过程的基本概念和应用领域以及在python中的实例

引言 高斯过程（Gaussian Process，简称GP）是一种概率模型，用于描述一组随机变量的联合概率分布，其中任何一个有限维度的子集都具有高斯分布 文章目录 引言一、高斯过程1.1 基本定义1.1.1 随机过程1.1.2 高斯分布 1.2 高斯过程的特性1.2.1 联合高斯性1.2.2 均值函数1.2.3 协方差函数（或核函数） 1.3 核函数1.4 高斯过程回归（Gauss
阅读更多...
Java ArrayList扩容机制 （源码解读）

Java ArrayList扩容机制 （源码解读）

结论：初始长度为10，若所需长度小于1.5倍原长度，则按照1.5倍扩容。若不够用则按照所需长度扩容。 一. 明确类内部重要变量含义         1：数组默认长度         2:这是一个共享的空数组实例，用于明确创建长度为0时的ArrayList ，比如通过 new ArrayList<>(0)，ArrayList 内部的数组 elementData 会指向这个 EMPTY_EL
阅读更多...
【编程底层思考】垃圾收集机制，GC算法，垃圾收集器类型概述

【编程底层思考】垃圾收集机制，GC算法，垃圾收集器类型概述

Java的垃圾收集（Garbage Collection，GC）机制是Java语言的一大特色，它负责自动管理内存的回收，释放不再使用的对象所占用的内存。以下是对Java垃圾收集机制的详细介绍： 一、垃圾收集机制概述： 对象存活判断：垃圾收集器定期检查堆内存中的对象，判断哪些对象是“垃圾”，即不再被任何引用链直接或间接引用的对象。内存回收：将判断为垃圾的对象占用的内存进行回收，以便重新使用。
阅读更多...
C++操作符重载实例(独立函数)

C++操作符重载实例(独立函数)

C++操作符重载实例，我们把坐标值CVector的加法进行重载，计算c3=c1+c2时，也就是计算x3=x1+x2，y3=y1+y2，今天我们以独立函数的方式重载操作符+（加号），以下是C++代码： c1802.cpp源代码： D:\YcjWork\CppTour>vim c1802.cpp #include <iostream>using namespace std;/*** 以独立函数
阅读更多...
【Tools】大模型中的自注意力机制

【Tools】大模型中的自注意力机制

摇来摇去摇碎点点的金黄 伸手牵来一片梦的霞光 南方的小巷推开多情的门窗 年轻和我们歌唱 摇来摇去摇着温柔的阳光 轻轻托起一件梦的衣裳 古老的都市每天都改变模样                      🎵 方芳《摇太阳》 自注意力机制（Self-Attention）是一种在Transformer等大模型中经常使用的注意力机制。该机制通过对输入序列中的每个元素计算与其他元素之间的相似性，
阅读更多...
如何通俗理解注意力机制？

如何通俗理解注意力机制？

1、注意力机制（Attention Mechanism）是机器学习和深度学习中一种模拟人类注意力的方法，用于提高模型在处理大量信息时的效率和效果。通俗地理解，它就像是在一堆信息中找到最重要的部分，把注意力集中在这些关键点上，从而更好地完成任务。以下是几个简单的比喻来帮助理解注意力机制： 2、寻找重点：想象一下，你在阅读一篇文章的时候，有些段落特别重要，你会特别注意这些段落，反复阅读，而对其他部分
阅读更多...
C++ | Leetcode C++题解之第393题UTF-8编码验证

C++ | Leetcode C++题解之第393题UTF-8编码验证

题目： 题解： class Solution {public:static const int MASK1 = 1 << 7;static const int MASK2 = (1 << 7) + (1 << 6);bool isValid(int num) {return (num & MASK2) == MASK1;}int getBytes(int num) {if ((num &
阅读更多...
实例：如何统计当前主机的连接状态和连接数

实例：如何统计当前主机的连接状态和连接数

统计当前主机的连接状态和连接数 在 Linux 中，可使用 ss 命令来查看主机的网络连接状态。以下是统计当前主机连接状态和连接主机数量的具体操作。 1. 统计当前主机的连接状态 使用 ss 命令结合 grep、cut、sort 和 uniq 命令来统计当前主机的 TCP 连接状态。 ss -nta | grep -v '^State' | cut -d " " -f 1 | sort |
阅读更多...
C语言 | Leetcode C语言题解之第393题UTF-8编码验证

C语言 | Leetcode C语言题解之第393题UTF-8编码验证

题目： 题解： static const int MASK1 = 1 << 7;static const int MASK2 = (1 << 7) + (1 << 6);bool isValid(int num) {return (num & MASK2) == MASK1;}int getBytes(int num) {if ((num & MASK1) == 0) {return
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2