[發(fā)明專利]一種流計算應(yīng)用中實現(xiàn)持久化的方法及裝置

在審

• 申請?zhí)枺?/td>: 201310362269.X
• 申請日: 2013-08-19
• 公開（公告）號: CN104424186A
• 公開（公告）日: 2015-03-18
• 發(fā)明（設(shè)計）人: 劉健男
• 申請（專利權(quán)）人: 阿里巴巴集團控股有限公司
• 主分類號: G06F17/30
• 分類號: G06F17/30
• 代理公司: 北京安信方達知識產(chǎn)權(quán)代理有限公司 11262
• 代理人: 解婷婷;栗若木

1.一種流計算應(yīng)用中實現(xiàn)持久化的方法，其特征在于，包括：

當(dāng)前批次消息消費成功，根據(jù)用于保存當(dāng)前正在消費的批次消息在消息隊列中的起始位置的第一起始偏移量和預(yù)先設(shè)置的持久化間隔，判斷是否需要進行持久化操作；

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

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

2.根據(jù)權(quán)利要求1所述的方法，其特征在于，所述流計算應(yīng)用正常啟動，或者故障恢復(fù)后啟動時，該方法還包括：

根據(jù)所述第二起始偏移量請求消息，同時將所述第一起始偏移量的值更改為第二起始偏移量的值。

3.根據(jù)權(quán)利要求2所述的方法，其特征在于，所述第二起始偏移量的值為空或者未保存有第二起始偏移量時，所述當(dāng)前批次消息位于消息中間件的消息隊列的起始位置；

同時還包括：設(shè)置所述第一起始偏移量的值為空。

4.根據(jù)權(quán)利要求1所述的方法，其特征在于，所述持久化操作失敗，該方法還包括：按照所述第一起始偏移量指示，重新對所述當(dāng)前批次消息中的消息進行消費。

5.根據(jù)權(quán)利要求2或4所述的方法，其特征在于，所述判斷是否需要進行持久化操作包括：將所述當(dāng)前批次的ID除以持久化間隔，當(dāng)其余數(shù)為零時，判斷出需要進行持久化操作；

其中，批次ID為從1開始以遞增步長為1的整數(shù)。

6.根據(jù)權(quán)利要求5所述的方法，其特征在于，所述流計算應(yīng)用正常啟動，或者故障恢復(fù)后啟動時，所述批次ID接著流計算應(yīng)用停止前的最后一個成功持久化過的批次ID繼續(xù)遞增步長為1遞增。

7.一種流計算應(yīng)用中實現(xiàn)持久化的裝置，其特征在于，至少存儲模塊、判斷模塊，以及處理模塊，其中，

存儲模塊，其中保存有持久化間隔、用于保存當(dāng)前正在消費的批次消息在消息隊列中的起始位置的第一起始偏移量，以及用于保存最近一次持久化操作的下一批次消息在消息隊列中的起始位置的第二起始偏移量；

判斷模塊，當(dāng)前批次消息消費成功，根據(jù)存儲模塊中保存的第一起始偏移量和預(yù)先設(shè)置的持久化間隔，判斷出需要進行持久化操作時向處理模塊發(fā)送持久化通知；

處理模塊，接收到來自判斷模塊的持久化通知，按照存儲模塊中保存的第二起始偏移量指示的消息位置進行持久化操作；并在持久化操作成功后，將第一起始偏移量和第二起始偏移量均更新為下一批次消息的起始偏移量。

8.根據(jù)權(quán)利要求7所述的裝置，其特征在于，所述處理模塊進一步用于：

在流計算應(yīng)用正常啟動，或者故障恢復(fù)后啟動時，根據(jù)所述存儲模塊中保存的第二起始偏移量從消息中間件中請求消息，同時將所述存儲模塊中保存的第一起始偏移量的值更改為第二起始偏移量的值。

9.根據(jù)權(quán)利要求7所述的裝置，其特征在于，所述處理模塊還進一步用于，在所述持久化操作失敗時，按照所述存儲模塊中保存的第一起始偏移量指示，重新對從消息中間件中當(dāng)前批次消息中的消息進行消費。

10.根據(jù)權(quán)利要求7～9任一項所述的裝置，其特征在于，所述判斷模塊具體用于：將所述第一起始偏移量所指示的當(dāng)前批次的ID除以所述持久化間隔，當(dāng)其余數(shù)為零時，判斷出需要進行持久化操作，向所述處理模塊發(fā)送持久化通知；其中，批次ID為從1開始以遞增步長為1的整數(shù)。