CN104424186B - 一种流计算应用中实现持久化的方法及装置 - Google Patents

Abstract

本申请公开了一种流计算应用中实现持久化的方法及装置，包括当前批次消息消费成功，根据第一起始偏移量和预先设置的持久化间隔，判断是否需要进行持久化操作；在需要进行持久化操作时，按照第二起始偏移量指示的消息位置进行持久化处理，并在持久化成功后，将第一起始偏移量和第二起始偏移量均更新为下一批次消息的起始偏移量。本申请中的持久化操作是在持久化间隔后进行的，增大了磁盘持久化的时间间隔，从而大大提高实时计算效率。在故障恢复时，最多只需要重新消费持久化间隔内的批次的消息，避免了现有同步持久化中频繁写磁盘带来的性能瓶颈，实时计算的消息吞吐性能提升了一个数量级；同时，将故障恢复带来的延迟缩短到了秒级，不会影响实时性。

Description

一种流计算应用中实现持久化的方法及装置

技术领域

本申请涉及流计算技术，尤指一种流计算应用中是实现持久化的方法及装置。

背景技术

通常，在流计算中将数据流称为消息，而对数据流的一系列计算、处理称为消费。

流计算产品主要用于实时计算。实时计算通常在内存中进行，而计算结果要通过某种途径保存并展现出来。目前，主要采用缓存或者持久化到磁盘如数据库（非内存数据库）中两种方式对计算结果进行保存。其中，由于缓存方式没有物理磁盘输入/输出（I/O），因此，缓存方式有着非常卓越的消息吞吐能力；但是，由于计算结果没有持久化，缓存方式几乎没有容错能力，也就是说，一旦出现应用程序中断、服务器宕机、缓存被清空等情况，保存在缓存中的计算结果将无法恢复。而采用持久化到磁盘中的方式，可以实现最高级别的容错功能，但是，持久化到磁盘涉及到大量的磁盘写，这又降低了流计算的计算速度，执行效率大约比采用缓存的方式低一个数量级。

图1为现有基本的容错型流计算应用中数据流向示意图，如图1所示，消息中间件集群发送的消息流是一条一条的。为了便于容错，通常流计算产品如图1中的流计算集群对消息流进行消费是以批次为单位的，即将若干条消息捆绑在一个批次中，每一个批次有一个唯一的标识（ID）。对于一个批次的消息，只有批次内的每一条消息都被成功消费后，这个批次的消息才被标记为被成功消费；只要一个批次内有一条消息没有被消费成功，整个批次的消息就会被消息中间件重新发送，被流计算集群重新消费。

最终处理过的消息流存储到磁盘称之为持久化，这一步对于故障恢复是至关重要的。一旦出现应用程序中断、服务器宕机等情况，只有持久化操作才能保证实时计算的结果不丢失。故障恢复、实时计算应用重启时，需要重新从磁盘中加载实时计算的过程数据与结果数据，将所有状态恢复到故障发生之前的一个正确的时间点。在图1所示的ZooKeeper集群中存储有消息中间件集群发送的消息所在的消息队列的偏移量。当消息中间件集群发送给流计算集群一批消息时，ZooKeeper中会记录这批次的消息在消息队列中的起始偏移量。如果这批次的消息被流计算集群成功消费，那么，消息中间件集群会发送下一批次消息，ZooKeeper中记录的偏移量随之更新为下一批次消息在消息队列中的起始偏移量；如果这批次消息被消费失败，那么，流计算集群会从ZooKeeper集群中重新读取这批次消息的偏移量，然后到消息中间件集群中重新请求该批次消息，以实现消息的失败重发。

发明内容

为了解决上述技术问题，本申请提供了一种流计算应用中实现持久化的方法及装置，能够保证故障后数据的安全恢复，提高实时计算效率。

为了达到本申请目的，本申请提供一种流计算应用中实现持久化的方法，包括：

当前批次消息消费成功，根据用于保存当前正在消费的批次消息在消息队列中的起始位置的第一起始偏移量和预先设置的持久化间隔，判断是否需要进行持久化操作；

在需要进行持久化操作时，按照用于保存最近一次持久化操作的下一批次消息在消息队列中的起始位置的第二起始偏移量指示的消息位置进行持久化处理；

持久化操作成功后，分别更新第一起始偏移量和第二起始偏移量为下一批次消息的起始偏移量。

所述流计算应用正常启动，或者故障恢复后启动时，该方法还包括：

根据所述第二起始偏移量请求消息，同时将所述第一起始偏移量的值更改为第二起始偏移量的值。

所述第二起始偏移量的值为空或者未保存有第二起始偏移量时，所述当前批次消息位于消息中间件的消息队列的起始位置；

同时还包括：设置所述第一起始偏移量的值为空。

所述持久化操作失败，该方法还包括：按照所述第一起始偏移量指示，重新对所述当前批次消息中的消息进行消费。

所述判断是否需要进行持久化操作包括：将所述当前批次的ID除以持久化间隔，当其余数为零时，判断出需要进行持久化操作；

其中，批次ID为从1开始以递增步长为1的整数。

所述流计算应用正常启动，或者故障恢复后启动时，所述批次ID接着流计算应用停止前的最后一个成功持久化过的批次ID继续递增步长为1递增。

本申请还公开一种流计算应用中实现持久化的装置，至少存储模块、判断模块，以及处理模块，其中，

存储模块，其中保存有持久化间隔、用于保存当前正在消费的批次消息在消息队列中的起始位置的第一起始偏移量，以及用于保存最近一次持久化操作的下一批次消息在消息队列中的起始位置的第二起始偏移量；

判断模块，当前批次消息消费成功，根据存储模块中保存的第一起始偏移量和预先设置的持久化间隔，判断出需要进行持久化操作时向处理模块发送持久化通知；

处理模块，接收到来自判断模块的持久化通知，按照存储模块中保存的第二起始偏移量指示的消息位置进行持久化操作；并在持久化操作成功后，将第一起始偏移量和第二起始偏移量均更新为下一批次消息的起始偏移量。

所述处理模块进一步用于：

在流计算应用正常启动，或者故障恢复后启动时，根据所述存储模块中保存的第二起始偏移量从消息中间件中请求消息，同时将所述存储模块中保存的第一起始偏移量的值更改为第二起始偏移量的值。

所述处理模块还进一步用于，在所述持久化操作失败时，按照所述存储模块中保存的第一起始偏移量指示，重新对从消息中间件中当前批次消息中的消息进行消费。

所述判断模块具体用于：将所述第一起始偏移量所指示的当前批次的ID除以所述持久化间隔，当其余数为零时，判断出需要进行持久化操作，向所述处理模块发送持久化通知；其中，批次ID为从1开始以递增步长为1的整数。

本申请提供的方案包括当前批次消息消费成功，根据用于保存当前正在消费的批次消息在消息队列中的起始位置的第一起始偏移量和预先设置的持久化间隔，判断是否需要进行持久化操作；在需要进行持久化操作时，按照用于保存最近一次持久化操作的下一批次消息在消息队列中的起始位置的第二起始偏移量指示的消息位置进行持久化处理，并在持久化成功后，将第一起始偏移量和第二起始偏移量均更新为下一批次消息的起始偏移量。本申请中的持久化操作不是针对每个批次成功消费后都要进行的，而是在一个预先设置的时间间隔即持久化间隔后进行的，增大了磁盘持久化的时间间隔，从而大大提高实时计算效率。这样，在故障恢复时，最多只需要重新消费持久化间隔内的批次的消息，与现有同步持久化相比，避免了同步持久化方案中频繁写磁盘带来的性能瓶颈，实时计算的消息吞吐性能提升了一个数量级，而且达到了与缓存方式的方案在同一个数量级；同时，将故障恢复带来的延迟缩短到了秒级，不会影响实时性。

本申请的其它特征和优点将在随后的说明书中阐述，并且，部分地从说明书中变得显而易见，或者通过实施本申请而了解。本申请的目的和其他优点可通过在说明书、权利要求书以及附图中所特别指出的结构来实现和获得。

附图说明

附图用来提供对本申请技术方案的进一步理解，并且构成说明书的一部分，与本申请的实施例一起用于解释本申请的技术方案，并不构成对本申请技术方案的限制。

图1为现有基本的容错型流计算应用中数据流向的示意图；

图2为本申请流计算应用中实现持久化的方法的流程图；

图3为本申请流计算应用中实现持久化的实施例的流程示意图；

图4为本申请流计算应用中实现持久化的装置的组成结构示意图。

具体实施方式

为使本申请的目的、技术方案和优点更加清楚明白，下文中将结合附图对本申请的实施例进行详细说明。需要说明的是，在不冲突的情况下，本申请中的实施例及实施例中的特征可以相互任意组合。

在本申请一个典型的配置中，计算设备包括一个或多个处理器（CPU）、输入/输出接口、网络接口和内存。

内存可能包括计算机可读介质中的非永久性存储器，随机存取存储器（RAM）和/或非易失性内存等形式，如只读存储器（ROM）或闪存（flashRAM）。内存是计算机可读介质的示例。

计算机可读介质包括永久性和非永久性、可移动和非可移动媒体可以由任何方法或技术来实现信息存储。信息可以是计算机可读指令、数据结构、程序的模块或其他数据。计算机的存储介质的例子包括，但不限于相变内存（PRAM）、静态随机存取存储器（SRAM）、动态随机存取存储器（DRAM）、其他类型的随机存取存储器（RAM）、只读存储器（ROM）、电可擦除可编程只读存储器（EEPROM）、快闪记忆体或其他内存技术、只读光盘只读存储器（CD-ROM）、数字多功能光盘（DVD）或其他光学存储、磁盒式磁带，磁带磁磁盘存储或其他磁性存储设备或任何其他非传输介质，可用于存储可以被计算设备访问的信息。按照本文中的界定，计算机可读介质不包括非暂存电脑可读媒体（transitory media），如调制的数据信号和载波。

在附图的流程图示出的步骤可以在诸如一组计算机可执行指令的计算机系统中执行。并且，虽然在流程图中示出了逻辑顺序，但是在某些情况下，可以以不同于此处的顺序执行所示出或描述的步骤。

目前，容错型流计算方案大致有以下几种：

一种是，将实时计算结果写入缓存，通过部署两套相同的计算集群来实现容灾。这种方式的优点是，由于没有磁盘I/O，执行效率快，并发访问性能高；但是缺点也是显而易见的，即部署成本翻倍，而且如果两套计算集群同时出现故障，计算结果仍然会丢失。

另一种是，启动一个实时计算应用以完成正常的业务计算功能，并将实时计算结果写缓存；同时，开启另外一个独立的实时计算应用以将接收到的原始消息持久化到磁盘中。由于持久化到磁盘的应用与完成业务计算的应用相互独立，因此，可以实现高执行效率。当出现集群宕机等故障时，在故障恢复时，需要对备份在磁盘中的消息进行重新消费，虽然不会造成数据丢失，但是，当消息量非常大时，故障恢复的过程导致了实时计算应用的严重延迟，从而失去了实时的价值。

还有一种是，在第二种方案的基础上进行改进，采用同步持久化的方法，即实时计算应用在消费每批次消息时，都将实时计算结果（不是原始消息，而是经过消费加工后的消息）持久化到磁盘中，而且，只有当对该批次的消费以及持久化操作全部成功后，消息中间件集群才会发送下一批次的消息。这种方式是目前通用的用于故障恢复的流计算方案。具体来讲，

同步持久化，就是每一个批次都要做一次写磁盘的操作。如图1所示，保存在ZooKeeper集群中的偏移量会随着对一个批次消息的成功消费而更新为下一个批次消息所在消息队列的起始位置。如果不做同步持久化，一旦发生应用中断、集群宕机等情况，在故障恢复、应用重启之后，将有部分数据丢失。举例来说，假设实时应用在成功消费完第T批次的消息后，没有进行持久化，将这批次的计算结果保存到磁盘中，而接着消费第(T+1)批次的消息，此时保存在Zookeeper集群中的偏移量已经更新为第(T+1)批次消息所在消息队列的起始位置。如果此时发生服务器宕机，在故障恢复、应用重启后，实时应用从Zookeeper集群中获得偏移量并到消息中间件集群中去请求消息，显然，这时请求到的是第(T+1)批次的消息；而之前第T批次的消息没有保存在磁盘中（即未执行同步持久化），必然导致第T批次消息的丢失。

在同步持久化方案中，由于每个批次都要做一次写磁盘的操作，而批次的发送时间间隔大约在400毫秒～2秒之间，实时应用对每个批次消息在内存中计算花费的时间大约在1秒钟以内。这样，磁盘写的频率会非常频繁。根据大量经验可以知道，每批次磁盘写花费的时间占整个批次消费所花费的总时间的比重达到了50%或更多，磁盘写成为了影响实时计算效率的主要瓶颈。

图2为本申请流计算应用中实现持久化的方法的流程图，如图2所示，包括以下步骤：

步骤200：当前批次消息消费成功，根据用于保存当前正在消费的批次消息在消息队列中的起始位置的第一起始偏移量和预先设置的持久化间隔，判断是否需要进行持久化操作。

本步骤中，对当前批次消息的消费属于现有技术，具体实现并不属于本申请的保护范围，这里不再赘述。

本步骤中，持久化间隔N是预先设置的一个大于1的整数，比如50。通常，可以将持久化间隔N设置为10到100之间的一个整数。

本步骤中的判断是否需要进行持久化操作包括：将当前批次的ID除以持久化间隔N，当其余数为零时，判断出需要进行持久化操作。其中，批次ID是从1开始递增步长为1的整数。

从本步骤可见，本申请中的持久化操作不是针对每个批次成功消费后都要进行的，而是在一个间隔即持久化间隔N后进行的。这样，在故障恢复时，最多只需要重新消费N个批次的消息，与现有同步持久化相比，避免了同步持久化方案中频繁写磁盘带来的性能瓶颈，实时计算的消息吞吐性能提升了一个数量级，而且达到了与不做持久化的方案如缓存方式在同一个数量级，同时，将故障恢复带来的延迟缩短到了秒级。

步骤201：在需要进行持久化操作时，按照用于保存最近一次持久化操作的下一批次消息在消息队列中的起始位置的第二起始偏移量指示的消息位置进行持久化处理。

持久化处理就是流计算应用从第二起始偏移量指示的消息位置，将数据缓冲中的计算结果数据写入到磁盘中。具体实现属于本领域技术人员的惯用技术手段，所不同的是，这里需要持久化的计算结果是由第二起始偏移量指示的，而第二起始偏移量保存的是最近一次持久化操作的下一批次消息在消息队列中的起始位置，也就是说，持久化的是从上一次持久化成功后的批次开始，包括持久化间隔N的批次消息在数据缓冲中的计算结果。即本申请中的持久化操作不是每个批次成功消费后都要进行的，而是在一个批次间隔即持久化间隔N的。

步骤202：持久化成功后，第一起始偏移量和第二起始偏移量均更新为下一批次消息的起始偏移量。本步骤的更新过程，保证了下一次持久化处理从未持久化处理的成功消费的批次消息开始。

在流计算应用正常启动，或者故障恢复后启动时，本申请方法还包括：根据第二起始偏移量请求消息，同时将第一起始偏移量的值更改为第二起始偏移量的值。此时，如果第二起始偏移量的值为空或者未保存有第二起始偏移量，则从消息中间件的消息队列的起始位置开始请求消息，同时第一起始偏移量的值设置为空。

同时，流计算应用正常启动，或者故障恢复后启动时，批次ID可以接着应用停止前的最后一个成功持久化过的批次ID继续递增，而不是重新从1开始递增。以保证批次ID对于同一个实时应用的唯一性。

如果持久化操作失败，本申请方法还包括：按照第一起始偏移量指示，重新对当前批次消息中的消息进行消费。

下面结合实施例对本发明方法进行详细描述。图3为本申请流计算应用中实现持久化的实施例的流程示意图，本实施例中以storm作为流计算框架，使用java语言开发流计算应用，并以图1所示容错型流计算应用为例进行描述，如图3所示，包括：

步骤300～步骤301：流计算应用启动如正常启动或故障恢复后启动，从ZooKeeper中读取第二起始偏移量，然后按照第二起始偏移量指示的消息位置到消息中间件中去请求消息，同时，将ZooKeeper中的第一起始偏移量的值更改为读取的第二起始偏移量的值。

本步骤中，如果ZooKeeper中没有保存第二起始偏移量或者其值为空，那么，从消息中间件的消息队列的起始位置开始请求消息，同时将ZooKeeper中的第一起始偏移量的值设置为空。

步骤302：到消息中间件中请求消息。

步骤303：流计算集群的计算单元对接收到的当前批次消息进行消费，如果消费失败，进入步骤308；否则进入步骤304。

步骤304：对当前批次消息的成功消费后，判断当前批次ID除以预先设置的持久化间隔N后得到的余数是否等于0。如果不等于0，说明此时还不需要进行持久化，进入步骤309；否则进入步骤305。

步骤305：在需要进行持久化操作时，进行持久化处理并将实时计算结果保存到磁盘中。

步骤306：判断持久化操作是否成功，如果成功进入步骤307；如果失败进入步骤310。其中，判断持久化是否成功属于本领域技术人员的惯用技术手段，通常数据库软件会提供给用户持久化的接口，调用接口后通过返回码来判断持久化是否成功即可。

步骤307：持久化操作成功，从消息中间件中获得下一批次消息的起始偏移量，同时将该偏移量保存为第一起始偏移量和第二起始偏移量，之后返回步骤302。

步骤308：如果对当前消息消费失败，从ZooKeeper中重新读取第一起始偏移量后返回步骤302。如果此时第一起始偏移量的值为空，则返回步骤302并从消息中间件的消息队列的起始位置开始请求消息。

步骤309：在不需要进行持久化操作时，从消息中间件中获取下一批次消息的起始偏移量，同时将该偏移量保存为第一起始偏移量后返回步骤302，继续对消息进行消费。

步骤310：如果持久化操作失败，从ZooKeeper中重新读取第一起始偏移量后返回步骤302。

图3所示流程在流计算应用接收到终止命令时停止。

图4为本申请流计算应用中实现持久化的装置的组成结构示意图，如图4所示，至少包括存储模块、判断模块，以及处理模块，其中，

存储模块，其中保存有持久化间隔、用于保存当前正在消费的批次消息在消息队列中的起始位置的第一起始偏移量，以及用于保存最近一次持久化操作的下一批次消息在消息队列中的起始位置的第二起始偏移量；

判断模块，当前批次消息消费成功，根据存储模块中保存的第一起始偏移量和预先设置的持久化间隔，判断出需要进行持久化操作时向处理模块发送持久化通知；具体用于，将第一起始偏移量所指示的当前批次的ID除以持久化间隔，当其余数为零时，判断出需要进行持久化操作；其中，批次ID为从1开始以递增步长为1的整数。

处理模块，接收到来自判断模块的持久化通知，按照存储模块中保存的第二起始偏移量指示的消息位置进行持久化处理；并在持久化成功后，将第一起始偏移量和第二起始偏移量均更新为下一批次消息的起始偏移量。

处理模块进一步用于：在流计算应用正常启动，或者故障恢复后启动时，根据存储模块中保存的第二起始偏移量从消息中间件中请求消息，同时将存储模块中保存的第一起始偏移量的值更改为第二起始偏移量的值。

处理模块还进一步用于，在持久化操作失败时，按照存储模块中保存的第一起始偏移量指示，重新对从消息中间件中当前批次消息中的消息进行消费。

以图1所示架构为例，本申请装置中的存储模块可以设置在ZooKeeper中，判断模块和处理模块可以设置在流计算集群中。在实际应用中，也中可以使用其他软件替代ZooKeeper，比如HBase、Mysql等；或者，将存储模块设置在消息中间件中实现等。

本领域的技术人员应该明白，上述的本申请实施例所提供的装置的各组成部分，以及方法中的各步骤，它们可以集中在单个的计算装置上，或者分布在多个计算装置所组成的网络上。可选地，它们可以用计算装置可执行的程序代码来实现。从而，可以将它们存储在存储装置中由计算装置来执行，或者将它们分别制作成各个集成电路模块，或者将它们中的多个模块或步骤制作成单个集成电路模块来实现。这样，本申请不限制于任何特定的硬件和软件结合。

虽然本申请所揭露的实施方式如上，但所述的内容仅为便于理解本申请而采用的实施方式，并非用以限定本申请。任何本申请所属领域内的技术人员，在不脱离本申请所揭露的精神和范围的前提下，可以在实施的形式及细节上进行任何的修改与变化，但本申请的专利保护范围，仍须以所附的权利要求书所界定的范围为准。

Claims (10)

1.一种流计算应用中实现持久化的方法，其特征在于，包括：

当前批次消息消费成功，根据用于保存当前正在消费的批次消息在消息队列中的起始位置的第一起始偏移量和预先设置的持久化间隔，判断是否需要进行持久化操作；

在需要进行持久化操作时，按照用于保存最近一次持久化操作的下一批次消息在消息队列中的起始位置的第二起始偏移量指示的消息位置进行持久化处理；

持久化操作成功后，分别更新第一起始偏移量和第二起始偏移量为下一批次消息的起始偏移量。

2.根据权利要求1所述的方法，其特征在于，所述流计算应用正常启动，或者故障恢复后启动时，该方法还包括：

根据所述第二起始偏移量请求消息，同时将所述第一起始偏移量的值更改为第二起始偏移量的值。

3.根据权利要求2所述的方法，其特征在于，所述第二起始偏移量的值为空或者未保存有第二起始偏移量时，所述当前批次消息位于消息中间件的消息队列的起始位置；

同时还包括：设置所述第一起始偏移量的值为空。

4.根据权利要求1所述的方法，其特征在于，所述持久化操作失败，该方法还包括：按照所述第一起始偏移量指示，重新对所述当前批次消息中的消息进行消费。

5.根据权利要求2或4所述的方法，其特征在于，所述判断是否需要进行持久化操作包括：将所述当前批次的ID除以持久化间隔，当其余数为零时，判断出需要进行持久化操作；

其中，批次ID为从1开始以递增步长为1的整数。

6.根据权利要求5所述的方法，其特征在于，所述流计算应用正常启动，或者故障恢复后启动时，所述批次ID接着流计算应用停止前的最后一个成功持久化过的批次ID继续递增步长为1递增。

7.一种流计算应用中实现持久化的装置，其特征在于，至少存储模块、判断模块，以及处理模块，其中，

存储模块，其中保存有持久化间隔、用于保存当前正在消费的批次消息在消息队列中的起始位置的第一起始偏移量，以及用于保存最近一次持久化操作的下一批次消息在消息队列中的起始位置的第二起始偏移量；

判断模块，当前批次消息消费成功，根据存储模块中保存的第一起始偏移量和预先设置的持久化间隔，判断出需要进行持久化操作时向处理模块发送持久化通知；

处理模块，接收到来自判断模块的持久化通知，按照存储模块中保存的第二起始偏移量指示的消息位置进行持久化操作；并在持久化操作成功后，将第一起始偏移量和第二起始偏移量均更新为下一批次消息的起始偏移量。

8.根据权利要求7所述的装置，其特征在于，所述处理模块进一步用于：

在流计算应用正常启动，或者故障恢复后启动时，根据所述存储模块中保存的第二起始偏移量从消息中间件中请求消息，同时将所述存储模块中保存的第一起始偏移量的值更改为第二起始偏移量的值。

9.根据权利要求7所述的装置，其特征在于，所述处理模块还进一步用于，在所述持久化操作失败时，按照所述存储模块中保存的第一起始偏移量指示，重新对从消息中间件中当前批次消息中的消息进行消费。

10.根据权利要求7～9任一项所述的装置，其特征在于，所述判断模块具体用于：将所述第一起始偏移量所指示的当前批次的ID除以所述持久化间隔，当其余数为零时，判断出需要进行持久化操作，向所述处理模块发送持久化通知；其中，批次ID为从1开始以递增步长为1的整数。