三、消息的可靠处理

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

相关文章

Python Transformers库(NLP处理库)案例代码讲解

《PythonTransformers库(NLP处理库)案例代码讲解》本文介绍transformers库的全面讲解,包含基础知识、高级用法、案例代码及学习路径,内容经过组织,适合不同阶段的学习者,对... 目录一、基础知识1. Transformers 库简介2. 安装与环境配置3. 快速上手示例二、核心模

一文详解Java异常处理你都了解哪些知识

《一文详解Java异常处理你都了解哪些知识》:本文主要介绍Java异常处理的相关资料,包括异常的分类、捕获和处理异常的语法、常见的异常类型以及自定义异常的实现,文中通过代码介绍的非常详细,需要的朋... 目录前言一、什么是异常二、异常的分类2.1 受检异常2.2 非受检异常三、异常处理的语法3.1 try-

Python使用getopt处理命令行参数示例解析(最佳实践)

《Python使用getopt处理命令行参数示例解析(最佳实践)》getopt模块是Python标准库中一个简单但强大的命令行参数处理工具,它特别适合那些需要快速实现基本命令行参数解析的场景,或者需要... 目录为什么需要处理命令行参数?getopt模块基础实际应用示例与其他参数处理方式的比较常见问http

Java Response返回值的最佳处理方案

《JavaResponse返回值的最佳处理方案》在开发Web应用程序时,我们经常需要通过HTTP请求从服务器获取响应数据,这些数据可以是JSON、XML、甚至是文件,本篇文章将详细解析Java中处理... 目录摘要概述核心问题:关键技术点:源码解析示例 1:使用HttpURLConnection获取Resp

Java中Switch Case多个条件处理方法举例

《Java中SwitchCase多个条件处理方法举例》Java中switch语句用于根据变量值执行不同代码块,适用于多个条件的处理,:本文主要介绍Java中SwitchCase多个条件处理的相... 目录前言基本语法处理多个条件示例1:合并相同代码的多个case示例2:通过字符串合并多个case进阶用法使用

Java实现优雅日期处理的方案详解

《Java实现优雅日期处理的方案详解》在我们的日常工作中,需要经常处理各种格式,各种类似的的日期或者时间,下面我们就来看看如何使用java处理这样的日期问题吧,感兴趣的小伙伴可以跟随小编一起学习一下... 目录前言一、日期的坑1.1 日期格式化陷阱1.2 时区转换二、优雅方案的进阶之路2.1 线程安全重构2

Redis消息队列实现异步秒杀功能

《Redis消息队列实现异步秒杀功能》在高并发场景下,为了提高秒杀业务的性能,可将部分工作交给Redis处理,并通过异步方式执行,Redis提供了多种数据结构来实现消息队列,总结三种,本文详细介绍Re... 目录1 Redis消息队列1.1 List 结构1.2 Pub/Sub 模式1.3 Stream 结

Python处理函数调用超时的四种方法

《Python处理函数调用超时的四种方法》在实际开发过程中,我们可能会遇到一些场景,需要对函数的执行时间进行限制,例如,当一个函数执行时间过长时,可能会导致程序卡顿、资源占用过高,因此,在某些情况下,... 目录前言func-timeout1. 安装 func-timeout2. 基本用法自定义进程subp

Java字符串处理全解析(String、StringBuilder与StringBuffer)

《Java字符串处理全解析(String、StringBuilder与StringBuffer)》:本文主要介绍Java字符串处理全解析(String、StringBuilder与StringBu... 目录Java字符串处理全解析:String、StringBuilder与StringBuffer一、St

浅析Java中如何优雅地处理null值

《浅析Java中如何优雅地处理null值》这篇文章主要为大家详细介绍了如何结合Lambda表达式和Optional,让Java更优雅地处理null值,感兴趣的小伙伴可以跟随小编一起学习一下... 目录场景 1:不为 null 则执行场景 2:不为 null 则返回,为 null 则返回特定值或抛出异常场景