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

2022-06-10 21:22:09

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机制以及源码分析

码农公寓

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

消息可靠性保障

元组状态

实现

FileReaderSpout

ContentStitchingBolt

数据源调整

spout

MessageWriterBolt

结语

相关文章