三、消息的可靠处理

2024-09-05 08:58
文章标签 处理 消息 可靠

本文主要是介绍三、消息的可靠处理,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

1、消息被完全处理的含义


当树创建完毕,并且树中的每一个消息都已经被处理时,Storm认为来自Spout的元组是“完全处理”的。当一个元组的消息树在指定的超时范围内不能被完全处理,则元组被认为是失败的。


2、如果一个消息被完全处理或完全处理失败会发生什么

首先,让我们看看Spout的元组的生命周期。ISpout接口的定义如下:

public interface ISpout extends Serializable {void open(Map conf, TopologyContext context, SpoutOutputCollector colle- ctor);void close();void nextTuple();void ack(Object msgId);void fail(Object msgId);
}

首先,Storm通过调用Spout的nextTuple()方法从Spout请求一个元组。Spout使用open()方法提供的SpoutOutputCollector对象发射一个元组到它的输出流。当发射元组时,Spout会提供一个“消息id”,以便用来识别元组。例如,KestrelSpout从Kestrel消息队列中读取一个消息时,会发射Kestrel提供的“消息id”。下面发射一个消息到SpoutOutputCollector对象:

_collector.emit(new Values("field1", "field2", 3) , msgId);

接下来,元组被发送到Bolt,同时Storm负责跟踪创建的消息树。如果Storm检测到一个元组是完全处理的,Storm将调用原Spout任务的ack()方法,把Spout提供给Storm的消息id作为输入参数。同样,如果元组超时,Storm将调用Spoutfail()方法。注意,一个元组将由Spout任务来确认成功或失败,这个Spout任务是创建此元组的完全相同的Spout任务。如果一个Spout跨集群执行很多任务,元组是不会被创建它的那个任务外的其他任务确认成功或失败的。

注意:bolt是没有ack()和fail()函数的,任何消息出错了,都是由根spout重发


3、Storm如何保证可靠性

在元组树中指定一个链接,此链接被称为锚定(Anchoring。Anchoring在发射一个新的元组的同一时间完成。让我们使用以下Bolt为例进行介绍,这个Bolt将包含一个句子的元组划分为一个包含每个单词的锚定:

public class SplitSentence extends BaseRichBolt {OutputCollector _collector;public void prepare(Map conf, TopologyContext context, OutputCollector 
collector) {_collector = collector;}public void execute(Tuple tuple) {String sentence = tuple.getString(0);for(String word: sentence.split(" ")) {_collector.emit(tuple, new Values(word));//锚定+发射}_collector.ack(tuple);//确认}public void declareOutputFields(OutputFieldsDeclarer declarer) {declarer.declare(new Fields("word"));}
}<

通过指定输入元组作为第一个参数来发射每个单词元组被锚定anchored。因为这个单词元组是被锚定的如果单词元组未能被下游处理,树的根的Spout元组将在稍后重发。相反,如果单词元组的发射操作如下,让我们看看会发生什么:

_collector.emit(new Values(word));

这种方式发射的单词元组导致未被锚定unanchored。如果元组未被下游处理,根元组将不会重发。这取决于你需要的Topology的容错保证,有时候需要相应地发射一个未被锚定的元组。

很多Bolt遵循一个读取一个输入元组,发射元组,在execute方法确认元组的通用模式。这些Bolt具有类别过滤器和简单的功能。Storm有一个接口称为BasicBolt,为你封装这个模式。SplitSentence的例子可以使用BasicBolt写成:

public class SplitSentence extends BaseBasicBolt {public void execute(Tuple tuple, BasicOutputCollector collector) {String sentence = tuple.getString(0);for(String word: sentence.split(" ")) {collector.emit(new Values(word));}}public void declareOutputFields(OutputFieldsDeclarer declarer) {declarer.declare(new Fields("word"));}        
}

这个实现与之前的实现相比,语义上是相同的,但是更简单。元组发射到BasicOutputCollectorare是自动Anchoring到输入元组execute方法完成时,输入元组自动为你确认。


4、Storm如何实现可靠性

       storm系统中有一组叫做acker(可以设置并行度为多个)的特殊任务,它负责跟踪DAG中的每个消息。每当发现一个DAG被完全处理,它就向创建这个根消息的Spout任务发送一个信号,该tuple tree已经被完全处理成功。

      系统使用一种哈希算法根据Spout消息的id来确定由哪个acker跟踪此消息派生出来的tuple tree。因为每个消息都知道与之对应的根消息的id(每当Bolt新生成一个消息,对应的tuple tree中的根消息id就复制到这个消息中),所以它知道应该与哪个acker通信。当这个消息被应答的时候,它就把关于tuple tree变化的信息发送给跟踪这棵树的acker。例如,它会告诉acker:“本消息已经处理完毕,但是我派生出来了一些新的消息,帮忙跟踪一下吧”

       一个Acker任务存储来自Spout元组id到一对值的映射。第一个值是创建Spout元组的任务id,通过这个ID,acker就知道消息处理完成时该通知哪个spout任务。第二个值是一个64位的值,称为ack val。它是所有消息的随机id的异或结果当一个Acker任务看到ack val已经成为0它就知道元组树已经完成了。



这篇关于三、消息的可靠处理的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1138509

相关文章

无人叉车3d激光slam多房间建图定位异常处理方案-墙体画线地图切分方案

墙体画线地图切分方案 针对问题:墙体两侧特征混淆误匹配,导致建图和定位偏差,表现为过门跳变、外月台走歪等 ·解决思路:预期的根治方案IGICP需要较长时间完成上线,先使用切分地图的工程化方案,即墙体两侧切分为不同地图,在某一侧只使用该侧地图进行定位 方案思路 切分原理:切分地图基于关键帧位置,而非点云。 理论基础:光照是直线的,一帧点云必定只能照射到墙的一侧,无法同时照到两侧实践考虑:关

【生成模型系列(初级)】嵌入(Embedding)方程——自然语言处理的数学灵魂【通俗理解】

【通俗理解】嵌入(Embedding)方程——自然语言处理的数学灵魂 关键词提炼 #嵌入方程 #自然语言处理 #词向量 #机器学习 #神经网络 #向量空间模型 #Siri #Google翻译 #AlexNet 第一节:嵌入方程的类比与核心概念【尽可能通俗】 嵌入方程可以被看作是自然语言处理中的“翻译机”,它将文本中的单词或短语转换成计算机能够理解的数学形式,即向量。 正如翻译机将一种语言

Thymeleaf:生成静态文件及异常处理java.lang.NoClassDefFoundError: ognl/PropertyAccessor

我们需要引入包: <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-thymeleaf</artifactId></dependency><dependency><groupId>org.springframework</groupId><artifactId>sp

ActiveMQ—消息特性(延迟和定时消息投递)

ActiveMQ消息特性:延迟和定时消息投递(Delay and Schedule Message Delivery) 转自:http://blog.csdn.net/kimmking/article/details/8443872 有时候我们不希望消息马上被broker投递出去,而是想要消息60秒以后发给消费者,或者我们想让消息没隔一定时间投递一次,一共投递指定的次数。。。 类似

jenkins 插件执行shell命令时,提示“Command not found”处理方法

首先提示找不到“Command not found,可能我们第一反应是查看目标机器是否已支持该命令,不过如果相信能找到这里来的朋友估计遇到的跟我一样,其实目标机器是没有问题的通过一些远程工具执行shell命令是可以执行。奇怪的就是通过jenkinsSSH插件无法执行,经一番折腾各种搜索发现是jenkins没有加载/etc/profile导致。 【解决办法】: 需要在jenkins调用shell脚

明明的随机数处理问题分析与解决方案

明明的随机数处理问题分析与解决方案 引言问题描述解决方案数据结构设计具体步骤伪代码C语言实现详细解释读取输入去重操作排序操作输出结果复杂度分析 引言 明明生成了N个1到500之间的随机整数,我们需要对这些整数进行处理,删去重复的数字,然后进行排序并输出结果。本文将详细讲解如何通过算法、数据结构以及C语言来解决这个问题。我们将会使用数组和哈希表来实现去重操作,再利用排序算法对结果

8. 自然语言处理中的深度学习:从词向量到BERT

引言 深度学习在自然语言处理(NLP)领域的应用极大地推动了语言理解和生成技术的发展。通过从词向量到预训练模型(如BERT)的演进,NLP技术在机器翻译、情感分析、问答系统等任务中取得了显著成果。本篇博文将探讨深度学习在NLP中的核心技术,包括词向量、序列模型(如RNN、LSTM),以及BERT等预训练模型的崛起及其实际应用。 1. 词向量的生成与应用 词向量(Word Embedding)

使用协程实现高并发的I/O处理

文章目录 1. 协程简介1.1 什么是协程?1.2 协程的特点1.3 Python 中的协程 2. 协程的基本概念2.1 事件循环2.2 协程函数2.3 Future 对象 3. 使用协程实现高并发的 I/O 处理3.1 网络请求3.2 文件读写 4. 实际应用场景4.1 网络爬虫4.2 文件处理 5. 性能分析5.1 上下文切换开销5.2 I/O 等待时间 6. 最佳实践6.1 使用 as

Level3 — PART 3 — 自然语言处理与文本分析

目录 自然语言处理概要 分词与词性标注 N-Gram 分词 分词及词性标注的难点 法则式分词法 全切分 FMM和BMM Bi-direction MM 优缺点 统计式分词法 N-Gram概率模型 HMM概率模型 词性标注(Part-of-Speech Tagging) HMM 文本挖掘概要 信息检索(Information Retrieval) 全文扫描 关键词

PHP7扩展开发之数组处理

前言 这次,我们将演示如何在PHP扩展中如何对数组进行处理。要实现的PHP代码如下: <?phpfunction array_concat ($arr, $prefix) {foreach($arr as $key => $val) {if (isset($prefix[$key]) && is_string($val) && is_string($prefix[$key])) {$arr[