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层面，来一步步说明。

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

本篇的代码存储在：

[email protected]: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， 当节点再度收到数据，便知道是否 处理。但依然要小心，在存储状态及发送数据的中间，拓扑挂掉，等等。

原文地址：https://www.cnblogs.com/zyzdisciple/p/10693336.html