你真的很熟分布式和事務嗎？

y35w 10年前發布 | 12K 次閱讀 事務

  微吐槽

看到滿世界的架構圖，心都碎了。

咱什么時候才能有架構的本事啊？

如果某招聘負責人給你推了個架構師，你一投，其立馬給又你推了一個高級XXX的職位讓你看，這不是玩我呢嗎...

簡歷上項目經歷寫一句hello world真的不行嘛？

不想了，我等碼農，還是看看怎么來處理分布式系統中的事務這個老大難吧！

本文略長，讀者需要有一定耐心，如果你是高級碼農或者架構師級別，你可以跳過。

本文注重實戰或者實現，不設計CAP，略提ACID。

本文適合基礎分布式程序員：

  1. 本文會涉及集群中節點的failover和recover問題；
  2. 本文會涉及事務及不透明事務的問題；
  3. 本文會提到微博和tweeter.
  4. 本文會引出一個大數據問題.

由于分布式這個話題太大，事務這個話題也太大，我們從一個集群的一個小小節點開始談起。

存活的節點與同步

分布式系統中，如何判斷一個節點（node）是否存活？kafka這樣認為：

  1. 此節點和zookeeper能喊話.（Keep sessions with zookeeper through heatbeats.）
  2. 此節點如果是個從節點，必須能夠盡可能忠實地反映主節點的數據變化。
    也就是說，必須能夠在主節點寫了新數據后，及時復制這些變化的數據，所謂及時，不能拉下太多哦.

那么，符合上面兩個條件的條件就可以認為是存活的，也可以認為是同步的（in-sync）.

關于第1點，大家對心跳都很熟悉，那么我們可以這樣認為某個節點不能和zookeeper喊話了： 

zookeeper-node:
var timer = 
new timer()
.setInterval(10sec)
.onTime(slave-nodes,function(slave-nodes){
    slave-nodes.forEach( node -> {
        boolean isAlive = node.heatbeatACK(15sec);
        if(!isAlive) {
            node.numNotAlive += 1;
            if(node.numNotAlive >= 3) {
                node.declareDeadOrFailed();
                slave-nodes.remove(node);

                //回調給leader-node-app?

            }
        }else 
        node.numNotAlive = 0;
    });
});

timer.run();

//你可以回調也可以像下面這樣簡單的計時判斷
leader-node-app:
var timer = 
new timer()
.setInterval(10sec)
.onTime(slave-nodes,function(slave-nodes){
    slave-nodes.forEach(node -> {
        if(node.isDeadOrFailed) {

        //node不能和zookeeper喊話了

        }
    });
});

timer.run();

關于第二點，要稍微復雜點了，這里也比較考究一個所謂大數據工程師或者架構師的功力了，怎么搞呢？來這么分析：

  • 數據 messages.
  • 操作 op-log.
  • 偏移 position/offset. 
// 1. 先考慮messages
// 2. 再考慮log的postion或者offset
// 3. 考慮msg和off都記錄在同源數據庫或者存儲設備上.(database or storage-device.)
var timer = 
new timer()
.setInterval(10sec)
.onTime(slave-nodes,function(nodes){
    var core-of-cpu = 8;
    //嫌慢就并發唄 mod hash go!
    nodes.groupParallel(core-of-cpu)
    .forEach(node -> {
        boolean nodeSucked = false;

        if(node.ackTimeDiff > 30sec) {
            //node卡住了
            nodeSucked = true;
        }
        if(node.logOffsetDiff > 10) {
            //node復制跟不上了
            nodeSucked = true;
        }

        if(nodeSucked) {
            //總之node死掉了
            node.declareDeadOrFailed();
            //不和你玩啦，集群不要你了
            nodes.remove(node);
            //該怎么處理呢，拋個事件吧.
            fire-event-NodeDeadOrFailed(node);
        }
    });
});

timer.run();

上面的節點的狀態管理一般由zookeeper來做，leader或者master節點也會維護那么點狀態。

那么應用中的leader或者master節點，只需要從zookeeper拉狀態就可以，同時，上面的實現是不是一定最佳呢？不是的，而且多數操作可以合起來，但為了描述節點是否存活這個事兒，咱們這么寫沒啥問題。

節點死掉、失敗、不同步了，咋處理呢？

好嘛，終于說到failover和recover了，那failover比較簡單，因為還有其它的slave節點在，不影響數據讀取。

  1. 同時多個slave節點失敗了？
    那...所以沒有100%的可用性，所以大家都是那么多個99.999999..所以愛奇藝才會掛，所以攜程...
  2. 如果主節點失敗了，那master-master不行嘛？
    keep-alived或者LVS或者你自己寫failover吧.
    高可用架構（HA）又是個大件兒了，此文不展開了。

我們來關注下recover方面的東西，這里把視野打開點，不僅關注slave節點重啟后追log來同步數據，我們看下在實際應用中，數據請求（包括讀、寫、更新）失敗怎么辦？

大家可能都會說，重試（retry）唄、重放（replay）唄或者干脆不管了唄！行，都行，這些都是策略，但具體怎么個搞法，你真的清楚了？

看看實際問題

我們先擺個探討的背景：

問題：消息流，比如微博的微博（真繞），源源不斷地流進我們的應用中，要處理這些消息，有個需求是這樣的：

某V發了一條微博（url），他的粉絲都看到這條微博了，那么認為這條微博(url)到達數為：粉絲數。

稱為Reach。

那么，統計一下3小時內的本條微博（url）的reach總數。

怎么解決呢？手段：

把某時間段內轉發過某條微博（url）的人拉出來，把這些人的粉絲拉出來，去掉重復的人，然后求總數，就是要求的reach.

為了簡單，我們忽略掉日期，先看看這個方法行不行： 

/** ---------------------------------
* 1. 求出轉發微博(url)的大V. 
* __________________________________*/

方法 ：getUrlToTweetersMap()

SQL ： /* 數據庫A，表url_user存儲了轉發某url的user */
SELECT url.user_id as tweeter_id
FROM url_user
WHERE url_user.url_id = ${url_id}

返回 ：[user_1,...,user_m]
/** ---------------------------------
* 2. 求出大V的粉絲 
* __________________________________*/

方法 : getFollowers(String tweeter_id);

SQL :   /* 數據庫B */
SELECT user.id as follower_id
FROM users
WHERE users.followee_id = ${tweeter_id}

返回：tweeter的粉絲
/** ---------------------------------
* 3. 求出Reach
* __________________________________*/

var tweeters = getUrlToTweetersMap();
var result = new HashMap<String,Integer>();
tweeters.forEach(t -> {
    // 你可以批量in + 并發讀來優化下面方法的性能
    var followers = getFollowers(t.tweeter_id);

    followers.forEach(f -> {
        //hash去重
        result.put(f.user_id,1);
    });
});

//Reach
return result.size();

頂呱呱，無論如何，求出了Reach啊！

其實這又引出了一個很重要的問題，也是很多大談框架、設計、模式卻往往忽視的問題：性能和數據庫建模的關系。

  1. 數據量有多大？
    在上面的數據庫設計中避免了JOIN，為了提高求大V粉絲的性能，可以將一批大V作為batch/bulk，然后多個batch并發讀，誓死搞死數據庫。
    這里將微博到轉發者表所在的庫，與粉絲庫分離，如果數據更大怎么辦？
    庫再分表...好吧。本文不從性能展開。
  2. 微博這種應用，人與人之間的關系成圖狀（網），你怎么建模存儲？而不僅僅對應這個問題，比如：
    某人的好友的好友可能和某人有幾分相熟？
  3. 你選擇MySQL分庫、表、區，還是選擇Redis，MongoDB作sharding？還是你直接就上HBase了？
  4. 二級索引在哪里？

其實很佩服一篇文章，幾個圖表，看起來很cool的架構圖就把這些問題說得一清二楚，架構嘛。

看看storm的解決方案： 

// url到大V
TridentState urlToTweeters =
    topology.newStaticState(getUrlToTweetersState());
// 大V到粉絲
TridentState tweetersToFollowers =
    topology.newStaticState(getTweeterToFollowersState());

topology.newDRPCStream("reach")
    .stateQuery(urlToTweeters, new Fields("args"), new MapGet(), new Fields("tweeters"))
    .each(new Fields("tweeters"), new ExpandList(), new Fields("tweeter"))
    .shuffle() /* 大V的粉絲很多，所以需要分布式處理*/
    .stateQuery(tweetersToFollowers, new Fields("tweeter"), new MapGet(), new Fields("followers"))
    .parallelismHint(200) /* 粉絲很多，所以需要高并發 */ 
    .each(new Fields("followers"), new ExpandList(), new Fields("follower"))
    .groupBy(new Fields("follower"))
    .aggregate(new One(), new Fields("one")) /* 去重 */
    .parallelismHint(20)
    .aggregate(new Count(), new Fields("reach")); /* 計算reach數 */

最多處理一次（At most once）

回到主題，引出上面的例子，一是為了引出一個可以分布式的話題，二是透漏這么點意思：很大、很空的東西誰都會說，誰都能聊，問題是寫的這個人真的懂不懂呢？有木有實現能力呢？

咱來務點實吧，前面我們說到 recover ，節點恢復的問題，那么我們恢復幾個東西？

  • 節點狀態
  • 節點數據

本篇從數據上來討論下這個問題，為使問題再簡單點，我們考慮寫數據的場景，如果我們用 write-ahead-log 的方式來保證數據復制和一致性，那么我們會怎么處理一致性問題呢？

  1. 主節點有新數據寫入.
  2. 從節點追log,準備復制這批新數據。從節點做兩件事：
    (1). 把數據的id偏移寫入log;
    (2). 正要處理數據本身，從節點掛了。

那么根據上文的節點存活條件，這個從節點掛了這件事被探測到了，從節點由維護人員手動或者其自己恢復了，那么在加入集群和小伙伴們繼續玩耍之前，它要同步自己的狀態和數據。問題來了：

如果根據log內的數據偏移來同步數據，那么，因為這個節點在處理數據之前就把偏移寫好了，可是那批數據lost-datas沒有得到處理，如果追log之后的數據來同步，那么那批數據lost-datas就丟了。

在這種情況下，就叫作數據最多處理一次，也就是說數據會丟失。

最少處理一次（At least once）

好吧，丟失數據不能容忍，那么我們換種方式來處理：

  1. 主節點有新數據寫入.
  2. 從節點追log,準備復制這批新數據。從節點做兩件事：
    (1). 先處理數據；
    (2). 正要把數據的id偏移寫入log，從節點掛了。

問題又來了：

如果從節點追log來同步數據，那么因為那批數據duplicated-datas被處理過了，而數據偏移沒有反映到log中，如果這樣追，會導致這批數據重復。

這種場景，從語義上來講，就是數據最少處理一次，意味著數據處理會重復。

僅處理一次（Exactly once）

Transaction

好吧，數據重復也不能容忍？要求挺高啊。

大家都追求的強一致性保證（這里是最終一致性），怎么來搞呢？

換句話說，在更新數據的時候，事務能力如何保障呢？

假設一批數據如下： 

// 新到數據
{
    transactionId:4
    urlId:99
    reach:5
}

現在要更新這批數據到庫里或者log里，那么原來的情況是： 

// 老數據
{
    transactionId：3
    urlId:99
    reach:3
}

如果說可以保證如下三點：

  1. 事務ID的生成是強有序的.（隔離性，串行）
  2. 同一個事務ID對應的一批數據相同.（冪等性，多次操作一個結果）
  3. 單條數據會且僅會出現在某批數據中.（一致性，無遺漏無重復）

那么，放心大膽的更新好了： 

// 更新后數據
{
    transactionId：4
    urlId:99
    //3 + 5 = 8
    reach:8
}

注意到這個更新是 ID偏移和數據一起更新 的，那么這個操作靠什么來保證： 原子性 。

你的數據庫不提供原子性？后文略有提及。

這里是更新成功了。如果更新的時候，節點掛了，那么庫里或者log里的id偏移不寫，數據也不處理，等節點恢復，就可以放心去同步，然后加入集群玩耍了。

所以說，要保證數據僅處理一次，還是挺困難的吧？

上面的保障“僅處理一次”這個語義的實現有什么問題呢？

性能問題。

這里已經使用了batch策略來減少到庫或磁盤的Round-Trip Time，那么這里的性能問題是什么呢？

考慮一下，采用master-master架構來保證主節點的可用性，但是一個主節點失敗了，另一個主節點當領導，是需要時間的。

假設從節點正在同步，啪！主節點掛了！因為要保證僅處理一次的語義，所以原子性發揮作用，失敗，回滾，然后從主節點拉失敗的數據（你不能就近更新，因為這批數據可能已經變化了，或者你根本沒緩存本批數據），結果是什么呢？

老主節點掛了， 新的主節點還沒啟動，所以這次事務就卡在這里，直到數據同步的源——主節點可以響應請求。

如果不考慮性能，就此作罷，這也不是什么大事。

你似乎意猶未盡？來吧，看看“銀彈”是什么？

Opaque-Transaction

現在，我們來追求這樣一種效果：

某條數據在一批數據中（這批數據對應著一個事務），很可能會失敗，但是它會在另一批數據中成功。換句話說，一批數據的事務ID一定相同。

來看看例子吧，老數據不變，只是多了個字段： prevReach 。 

// 老數據
{
    transactionId：3
    urlId:99
    //注意這里多了個字段，表示之前的reach的值
    prevReach:2
    reach:3
}


// 新到數據
{
    transactionId:4
    urlId:99
    reach:5
}

這種情況，新事務的ID更大、更靠后，表明新事務可以執行，還等什么，直接更新，更新后數據如下： 

// 新到數據
{
    transactionId:4
    urlId:99
    //注意這里更新為之前的值
    prevReach:3
    //3 + 5 = 8
    reach:8
}

現在來看下另外的情況： 

// 老數據
{
    transactionId：3
    urlId:99
    prevReach:2
    reach:3
}

// 新到數據
{
    //注意事務ID為3，和老數據中的事務ID相同
    transactionId:3
    urlId:99
    reach:5
}

這種情況怎么處理？是跳過嗎？因為新數據的事務ID和庫里或者log里的事務ID相同，按事務要求這次數據應該已經處理過了，跳過？不，這種事不能靠猜的，想想我們有的幾個性質，其中關鍵一點就是：

給定一批數據，它們所屬的事務ID相同。

仔細體會下，上面那句話和下面這句話的差別：

給定一個事務ID，任何時候，其所關聯的那批數據相同。

我們應該這么做，考慮到新到數據的事務ID和存儲中的事務ID一致，所以這批數據可能被分別或者異步處理了，但是，這批數據對應的事務ID永遠是同一個，那么，即使這批數據中的A部分先處理了，由于大家都是一個事務ID，那么A部分的前值是可靠的。

所以，我們將 依靠prevReach而不是Reach的值 來更新： 

// 更新后數據
{
    transactionId:3
    urlId:99
    //這個值不變
    prevReach:2
    //2 + 5 = 7
    reach:7
}

你發現了什么呢？不同的事務ID，導致了不同的值：

  1. 當事務ID為4，大于存儲中的事務ID3，Reach更新為3+5 = 8.
  2. 當事務ID為3，等于存儲中的事務ID3，Reach更新為2+5 = 7.

這就是 Opaque Transaction .

這種事務能力是最強的了，可以保證事務異步提交。所以不用擔心被卡住了，如果說集群中：

Transaction：

  • 數據是分批處理的，每個事務ID對應一批確定、相同的數據.
  • 保證事務ID的產生是強有序的.
  • 保證分批的數據不重復、不遺漏.
  • 如果事務失敗，數據源丟失，那么后續事務就卡住直到數據源恢復.

Opaque-Transaction：

  • 數據是分批處理的，每批數據有確定而唯一的事務ID.
  • 保證事務ID的產生是強有序的.
  • 保證分批的數據不重復、不遺漏.
  • 如果事務失敗，數據源丟失，不影響后續事務，除非后續事務的數據源也丟了.

其實這個全局ID的設計也是門藝術：

  • 冗余關聯表的ID，以減少join，做到O(1)取ID.
  • 冗余日期（long型）字段，以避免order by.
  • 冗余過濾字段，以避免無二級索引（HBase）的尷尬.
  • 存儲mod-hash的值，以方便分庫、分表后，應用層的數據路由書寫.

這個內容也太多，話題也太大，就不在此展開了。

你現在知道推ter的snowflake生成全局唯一且有序的ID的重要性了。

兩階段提交

現在用zookeeper來做兩階段提交已經是入門級技術，所以也不展開了。如果不打算使用上面的強有序事務ID技術，那么就考慮兩階段提交吧。

如果你的數據庫不支持原子操作，那么也考慮兩階段提交吧。

結語

現在，你真的懂分布式和事務嗎？

原文  http://www.cnblogs.com/foreach-break/p/distributed_system_and_trans

 本文由用戶 y35w 自行上傳分享，僅供網友學習交流。所有權歸原作者，若您的權利被侵害，請聯系管理員。
 轉載本站原創文章，請注明出處，并保留原始鏈接、圖片水印。
 本站是一個以用戶分享為主的開源技術平臺，歡迎各類分享！
  本文地址：http://www.baiduhome.net/news/view/1a2e074