Storm系列三: Storm消息可靠性保障

时间:2022-08-30 13:40:52

Storm系列三: Storm消息可靠性保障

在上一篇 Storm系列二: Storm拓扑设计 中我们已经设计了一个稍微复杂一点的拓扑。

而本篇就是在上一篇的基础上再做出一定的调整。

在这里先大概提一下上一篇的业务逻辑, 我们会不断收到来自前端的消息,消息包含消息的发送时间,消息内容,结束标识, 消息的发送者, SessionId等其他信息, 我们需要做的事情是当接收到消息之后,根据SessionId判断是否属于同一消息, 如果是的话将内容拼接, 如果结束标识为 true, 表示会话已结束,则存入数据库或其他地方, 如果不为true, 则等待, 在1分钟后 还是没有收到消息, 则存入数据库。

在上一篇中, 消息内容指示的是用户行为, 因此对于消息的可靠性保障并没有要求。

现在我们将需求微调,消息内容是系统的日志信息, 并保证日志信息没有遗漏, 因为很可能在将来我们需要查找到系统的日志消息, 判定某些错误发生的原因,就要保证毫无遗漏。

那么什么是消息可靠性呢?

如果我们的拓扑因为某种意外终止了,当拓扑再度恢复,总不可能从头开始读取数据,又或者数据因为时效性已经丢失,无法被再度获取, 所以首先,我们要有一个可靠的数据源。 这意味着需要有保存数据的能力, 除非通知数据已经被消费,否则就不能删除数据, 同时要记录每次消费到某个位置。

当有了可靠的数据源之后,因为故障意外,某个bolt所处的节点挂掉,导致正在处理的数据被丢弃了, 所以需要Spout再度发送数据。 那么第二点,需要一个有重发指定消息的能力。

我们已经满足了上述两点,那么Spout是如何得知当前数据已经被处理掉了呢? 无论是成功还是失败,总需要通过某种途径获取其监听信息。 所以第三点就是,需要一个能够被一路跟踪状态信息的元组流。 也叫做锚定。因为下游并不止一个bolt,可能会在任何一个节点出问题, 所以需要持续跟踪。

第四点是, 需要一个具有容错能力的Storm拓扑。

当然我们能够影响的只有前三点。

在这里选择一个可靠的数据源,文本输入当然是可以的, 实际中使用的是 kafka, RabbitMQ, 等消息队列, 作为可靠数据源的输入。

消息可靠性保障

在Storm中的消息可靠性保障意味着, 消息以元组的形式从spout中发射出来,并经过拓扑中的各个bolt完成处理, 如果一个元组在某一个节点处理失败, Storm会立刻得知相信的信息,并通知Spout可以进行相应的处理,无论是重发还是抛弃(当然,如果是抛弃的话, 这里并没有必要采取可靠数据源, 也不怎么需要可靠性保障。因为Spout表现出的行为是对元组的成功失败漠不关心。),直到,这个元组被完成掉。

元组状态

元组有两种状态,ack 和 fail,当Spout发射出一个元组之后,下游的bolt在处理完成之后,可能会发出更多的元组, Storm为spout发出的每一个tuple都创建了一个元组树(tupletree), 其中Spout发射的元组成为根元组,当一棵元组树的所有叶节点都完成了对元组的处理,此时storm才会认为 对当前tuple已完成处理。

那么storm中的消息保障本身就是可选的, 你可能在任意节点决定, 当前元组已经完成, 后续的所有处理并不需要再度进行保障性操作。既然*度是这样高, 因此你需要做这样两件事:

  • 在每个节点发出元组的时候进行 锚定,也就是意味着storm需要继续跟踪当前元组的状态。

  • 确保你的每一个叶节点会对tuple进行应答,告诉storm我已经处理完成了。

当然,没有人能保证bolt永远会做出应答,即使bolt挂掉了,storm依然会跟踪元组状态,在得不到回应的情况下,元组树将会报错, 表示当前tuple处于fail状态, 这个时间的配置是TOPOLOGY_MESSAGE_TIMEOUT_SECS, 缺省配置为30s。

Config conf = new Config();
conf.put(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS, 30);

实现

已经有了相应的概念, 下面的部分我们开始从spout层面,来一步步说明。

代码基于上一篇中的代码进行修改, 开头已经提到过了。

本篇的代码存储在:

git@github.com:zyzdisciple/storm_study.git

的guaranteed_message包下。

需要提到的一点是, 我继承的都是 BaseRichSpout/BaseRichBolt;

在BaseBasic系列中, 已经默认为你做了相关处理。

FileReaderSpout

if (isValid(info, line)) {
completingMessage(info);
//判断当前的时间区间.
long timeGroup = System.currentTimeMillis() / (BehaviorConstants.SESSION_TIME_OUT_SECS * 1000);
collector.emit(new Values(timeGroup, info), info.hashCode());
}

在这里,核心在:

collector.emit(new Values(timeGroup, info), info.hashCode());

collector.emit的第二个参数配置,就是 msgId, 如果我们在从Spout中发射数据时, 没有配置messageId,那么storm并不会跟踪元组状态, 即使后续再怎么处理, 也是无效的。

在这里, 我简单的采取了hashCode, 当然也重写了info的hashCode方法, 在实际中, 我们从kafka数据源中拉取数据的时候, 一般都会有其ID作为唯一性标识, 并不需要去单独创建。

其他工作暂且不提, 让我们继续。

ContentStitchingBolt

更改的有这样一个个地方:

if (info.getEnd()) {
collector.emit(input, new Values(info));
//发送后需要移除相关数据
collectMap.remove(key);
} else {
collectMap.put(key, info);
}

会发现在 emit的同时,在第一个emit中传递了 当前tuple作为参数,这就是进行了锚定行为, 将spout发出的tuple与后续的相关联, 可以监听状态, 如果不监听而直接响应ack,那么系统会认为你已经完成了, 如果不监听也不响应,时间到了,系统会认为你超时了。

对于系统消息我们并没有进行 ack处理, 这是因为storm仅跟踪 spout发出的tuple, 对于系统消息, 并不需要理会。

然而,在数据不满足直接发射条件的时候, 我们对tuple并没有进行ack,考虑如果ack,表示tuple在当前节点已经完成处理,如果不存在后续bolt的话, 则可以认为整个tuple都已经处理完毕, 那么在spout中就会删除对应数据, 基于可靠数据源也会忽略该数据, 然而事实上目前的数据我们是存在内存中的,当bolt挂掉, 则内存相关数据消息, 那么就真的是完全无法恢复了。

而不进行ack, 那么就会出现这样一个问题, 当我们的定时器,messageTimeout超时之前, tuple的定时器已经超时了,此时会重新发出一条数据,造成了更多的困扰, 所以必须有这样一条要求:

我们的tuple超时时间必须大于 messageTimeout。

但这样就不会造成问题了吗? 并不是。

如果我们的messageTimeout设置的时间本来就很长,比如十分钟, 那么tupleTimeout必须大于十分钟,也就是一条tuple发出去之后, 十分钟我才能将其定义为失败状态, 这没什么, 但是十分钟内会有多少条数据累积?

因此另一条配置也是比较有用的:

Config conf = new Config();
conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 30000);

这一条同样是配置给topology级别的:



表示的意思是, 如果在整个拓扑中, 有超过30000条tuple处于未响应状态, 那么spout就停止发送数据, 将其阻塞掉.

但仔细思考会发现仅仅这样处理是不够的,假定存在数据属于同一SessionId:

1,2,3,4,5

按照目前的假设来看,我们收到1234时既不进行ack,也不锚定,唯有收到5的时候再做处理, 此时是否应该取出1234所属的tuple一一ack?并不合适,理应需要对1234再度进行锚定,因为唯有下游有权决定数据到底是处理失败了还是成功了。那我们对1234再进行锚定发送? 也不合适,因为这意味着要将数据发送5遍,有4条数据是完全无效的。

那么首先可以确定的是,当收到中间数据需要进行ack,当真正需要发送数据的时候再进行锚定,也就是收到5的时候进行锚定,锚定的对象又是谁呢?是1所在的tuple。

那再来分析一下,身为数据源Spout,需要满足怎样的特性才能够保证收到2345的时候并不删除数据,只有收到1的时候,再将数据删除掉。

也就是说,数据源是一个队列,唯有当收到第一个数据的ack时,才按顺序检测,一一删除,否则都不删除。

这里我们做的是简化处理,毕竟真正的可靠消息,有kafka这些专门的消息组件进行保证。

分析了这么久终于可以开始代码了。

而在这之前,小小的总结一下:

  • 唯有当数据确定不需要再度进行回放,一是数据已经被彻底处理掉了,没有利用价值,二是保存在了另一种可靠的存储结构中, 此时我们才能进行ack,通知数据源,数据已经无效了。

  • 作为数据源也需要为消息的可靠性提供一定的保障, 不能够跨节点删除, 最好是只能够按序删除, 进行标记删除的处理方式。

数据源调整

import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicLong; /**
* 自己设计的数据源, 需要完成一系列功能。
* @author zyzdisciple
* @date 2019/4/11
*/
public enum DataSource { INSTANCE; private BufferedReader br; private BlockingQueue<Node> queue; private AtomicLong seq; private BlockingDeque<Long> ackIndexes; private static final Object deleteQueueLock = new Object(); private static final Logger logger = LoggerFactory.getLogger(DataSource.class); DataSource() {
try {
br = new BufferedReader(new FileReader("E:\\IdeaProjects\\storm_demo\\src\\main\\resources\\user_behavior_data.txt"));
} catch (FileNotFoundException e) {
e.printStackTrace();
}
queue = new LinkedBlockingDeque<>();
ackIndexes = new LinkedBlockingDeque<>();
} /**
* 获取一行数据,没有返回null。
* @return
*/
public String nextLine() {
Node node = null;
try {
String line = br.readLine();
if (!line.trim().isEmpty()) {
node = new Node(seq.getAndIncrement(), line);
queue.add(node);
}
} catch (IOException e) {
logger.warn("empty queue, e:" + e);
}
return node == null ? null : node.getValue();
} /**
* 成功响应
* @param seq
*/
public void ack(Object seq) {
if (seq == null) {
return;
}
deleteNode(Long.parseLong(seq.toString()));
} private void deleteNode(long seq) {
synchronized (deleteQueueLock) {
Node headNode = queue.peek();
if (headNode != null && headNode.getIndex() == seq) {
queue.poll();
//一直向下删除, 直到不等
deepDelete();
} else {
Long headIndex = ackIndexes.peek();
if (headIndex == null) {
ackIndexes.add(seq);
} else if (seq > headIndex) {
ackIndexes.addLast(seq);
} else if (seq < headIndex) {
ackIndexes.addFirst(seq);
}
}
}
} /**
* 继续向下删除
*/
private void deepDelete() {
Node headNode = queue.peek();
long seq = ackIndexes.peek();
boolean hasDeleted = false;
if (headNode != null && headNode.getIndex() == seq) {
queue.poll();
ackIndexes.poll();
//一直向下删除, 直到不等
deepDelete();
}
}
}

其主要有两个功能, 一个是ack,删除缓存数据, 另一个是取出数据。

spout

@Override
public void nextTuple() {
Node node = dataSource.nextLine();
if (node == null) {
return;
}
String line = node.getValue();
MessageInfo info = gson.fromJson(line, MessageInfo.class);
if (isValid(info, line)) {
completingMessage(info);
//判断当前的时间区间.
long timeGroup = System.currentTimeMillis() / (BehaviorConstants.SESSION_TIME_OUT_SECS * 1000);
collector.emit(new Values(timeGroup, info), node.getIndex());
}
}

主要是改写了nextTuple, 在这里用我们的“可靠数据源”接收数据,进行响应。

在这里会发现一点特性,贯穿数据源-spout-bolt的整个过程,其中key,也就是我们定义的ID,起到了桥梁的作用

那么当数据真正处理完成,收到下游的ack之后,又应该作何处理?这就需要关注Spout的接口了。 我们会注意到, 还提供了这样一个方法:

void ack(Object msgId);

因此我们重写这个方法就可以了,告诉dataSource当前数据已经处理完了。

@Override
public void ack(Object msgId) {
dataSource.ack(msgId);
}

那么如果下有数据处理失败了,自然有另一个方法,fail:

void fail(Object msgId);

那么当我们接收到fail时, 该从dataSource... 等等,dataSource并没有提供根据msgId取出对应数据的功能啊,是我们疏忽忘记了吗?并不是,消息队列,是一个队列,并不支持根据msgId查询返回特定的数据, 大多数情况下, 我们都需要自己维护相应的数据。

//加入属性cacheMap
private Map<Long, MessageInfo> cacheMap;
//在open方法中初始化
cacheMap = new HashMap<>();
//在nextTuple发射之前
cacheMap.put(node.getIndex(), info);
//在ack中收到消息之后
cacheMap.remove(msgId);
//在fail中
@Override
public void fail(Object msgId) {
long timeGroup = System.currentTimeMillis() / (BehaviorConstants.SESSION_TIME_OUT_SECS * 1000);
collector.emit(new Values(timeGroup, cacheMap.get(msgId)), msgId);
}

这样就完成了一个tuple从生到死的过程处理。

等等,在上一篇我们提到过一个问题,spout一样是可以设置并行度的,也就是说可能会存在多个线程,我们这么操作cacheMap并且不加锁真的好吗?

内容处理bolt就不贴在这里了,设计仍有一定不合理的地方, 但已经能说明主要问题。外加代码太多, 有兴趣可以自己去github上看一下。

MessageWriterBolt

@Override
public void execute(Tuple input) {
MessageInfo info = (MessageInfo) input.getValueByField(BehaviorConstants.FIELD_INFO);
String jsonMessage = null;
try {
jsonMessage = gson.toJson(info, MessageInfo.class);
} catch (Exception e) {
logger.warn("格式转换失败, e" + e);
collector.ack(input);
}
try {
pw.println(jsonMessage);
pw.flush();
} catch (Exception e) {
logger.error("写入文件失败, e:" + e);
collector.fail(input);
}
}

在这里,对于两种不同的情况会看到我们的 ack ,fail处理方式有所不同, 为什么呢?

对于错误我们分为,已知的和未知的, 对于已知的错误,也有两种,可重试,和不可重试,对于不可重试错误,数据错了就错了,再试一千次也是错的,所以直接响应ack。

对于可重试错误,如数据库插入失败等其他情况,就可以告知拓扑失败信息,促使重试。

而对于未知错误,那自然是没办法处理了,只能等到它发生变成已知错误,再处理。

结语

在本章主要讲了数据的可靠性保障相关的东西, 了解了实现可靠性的基本要求是, 一个可靠的数据源, 一个锚定的元组流, 一个能够感知并处理元组状态的spout。

还有很重要的一点没有提到, 是 一个容错性的拓扑。

概念比较宽泛,需要考虑到整个拓扑如果挂掉,如何恢复数据,从上次的某个地方继续向下读取数据, 如果某个bolt挂掉,相应的数据ack相关又该怎样处理, 以及与外界交互的,如文件流,数据库写入等等地方, 出现问题又该怎样处理?

同时,对于数据处理可靠性级别有这样几种:

最多一次, 至少一次, 仅一次。

最多一次就意味着可以不处理数据, 不可靠的数据源就是这样的

至少一次,只要我们能够对拓扑中的 ack,fail使用的谨慎而明白,这一点也是很好保证的。

仅一次, 如果我们在处理的是扣费项目, 因为数据重新发送,导致重复扣费,别人会投诉你的。 所以需要对数据加入唯一性标识, 并且将数据的处理状态, 处理节点等等都交给另一个可靠的系统进行维护。

在自己设计的时候,有这样一个简单的处理办法:

对于我们处理的每一个节点,举个例子:

有8个流程需要执行,顺序未必一致。

我们只需要始终在一个可靠的地方,维护数据状态:

000 000 00 8位0,当某个节点被处理,即置为1, 当节点再度收到数据,便知道是否 处理。但依然要小心,在存储状态及发送数据的中间,拓扑挂掉,等等。

关于Storm中Ack的详细机制:Apache Storm 实时流处理系统ACK机制以及源码分析