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

在審

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

技術領域

本申請涉及流計算技術，尤指一種流計算應用中是實現(xiàn)持久化的方法及裝置。

背景技術

通常，在流計算中將數(shù)據(jù)流稱為消息，而對數(shù)據(jù)流的一系列計算、處理稱為消費。

流計算產(chǎn)品主要用于實時計算。實時計算通常在內(nèi)存中進行，而計算結果要通過某種途徑保存并展現(xiàn)出來。目前，主要采用緩存或者持久化到磁盤如數(shù)據(jù)庫（非內(nèi)存數(shù)據(jù)庫）中兩種方式對計算結果進行保存。其中，由于緩存方式?jīng)]有物理磁盤輸入/輸出（I/O），因此，緩存方式有著非常卓越的消息吞吐能力；但是，由于計算結果沒有持久化，緩存方式幾乎沒有容錯能力，也就是說，一旦出現(xiàn)應用程序中斷、服務器宕機、緩存被清空等情況，保存在緩存中的計算結果將無法恢復。而采用持久化到磁盤中的方式，可以實現(xiàn)最高級別的容錯功能，但是，持久化到磁盤涉及到大量的磁盤寫，這又降低了流計算的計算速度，執(zhí)行效率大約比采用緩存的方式低一個數(shù)量級。

圖1為現(xiàn)有基本的容錯型流計算應用中數(shù)據(jù)流向示意圖，如圖1所示，消息中間件集群發(fā)送的消息流是一條一條的。為了便于容錯，通常流計算產(chǎn)品如圖1中的流計算集群對消息流進行消費是以批次為單位的，即將若干條消息捆綁在一個批次中，每一個批次有一個唯一的標識（ID）。對于一個批次的消息，只有批次內(nèi)的每一條消息都被成功消費后，這個批次的消息才被標記為被成功消費；只要一個批次內(nèi)有一條消息沒有被消費成功，整個批次的消息就會被消息中間件重新發(fā)送，被流計算集群重新消費。

最終處理過的消息流存儲到磁盤稱之為持久化，這一步對于故障恢復是至關重要的。一旦出現(xiàn)應用程序中斷、服務器宕機等情況，只有持久化操作才能保證實時計算的結果不丟失。故障恢復、實時計算應用重啟時，需要重新從磁盤中加載實時計算的過程數(shù)據(jù)與結果數(shù)據(jù)，將所有狀態(tài)恢復到故障發(fā)生之前的一個正確的時間點。在圖1所示的ZooKeeper集群中存儲有消息中間件集群發(fā)送的消息所在的消息隊列的偏移量。當消息中間件集群發(fā)送給流計算集群一批消息時，ZooKeeper中會記錄這批次的消息在消息隊列中的起始偏移量。如果這批次的消息被流計算集群成功消費，那么，消息中間件集群會發(fā)送下一批次消息，ZooKeeper中記錄的偏移量隨之更新為下一批次消息在消息隊列中的起始偏移量；如果這批次消息被消費失敗，那么，流計算集群會從ZooKeeper集群中重新讀取這批次消息的偏移量，然后到消息中間件集群中重新請求該批次消息，以實現(xiàn)消息的失敗重發(fā)。

發(fā)明內(nèi)容

為了解決上述技術問題，本申請?zhí)峁┝艘环N流計算應用中實現(xiàn)持久化的方法及裝置，能夠保證故障后數(shù)據(jù)的安全恢復，提高實時計算效率。

為了達到本申請目的，本申請?zhí)峁┮环N流計算應用中實現(xiàn)持久化的方法，包括：

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

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

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

所述流計算應用正常啟動，或者故障恢復后啟動時，該方法還包括：

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

所述第二起始偏移量的值為空或者未保存有第二起始偏移量時，所述當前批次消息位于消息中間件的消息隊列的起始位置；

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

所述持久化操作失敗，該方法還包括：按照所述第一起始偏移量指示，重新對所述當前批次消息中的消息進行消費。

所述判斷是否需要進行持久化操作包括：將所述當前批次的ID除以持久化間隔，當其余數(shù)為零時，判斷出需要進行持久化操作；

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

所述流計算應用正常啟動，或者故障恢復后啟動時，所述批次ID接著流計算應用停止前的最后一個成功持久化過的批次ID繼續(xù)遞增步長為1遞增。

本申請還公開一種流計算應用中實現(xiàn)持久化的裝置，至少存儲模塊、判斷模塊，以及處理模塊，其中，

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