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

本文注重實戰或者實現，不涉及CAP，略提ACID。
本文適合基礎分佈式程序員：

一、本文會涉及集羣中節點的failover和recover問題.
二、本文會涉及事務及不透明事務的問題.
三、本文會提到微博和tweeter，並引出一個大數據問題.
因爲分佈式這個話題太大，事務這個話題也太大，咱們從一個集羣的一個小小節點開始談起。

集羣中存活的節點與同步

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

一、此節點和zookeeper能喊話.（Keep sessions with zookeeper through heartbeats.）
二、此節點若是是個從節點，必須可以儘量忠實地反映主節點的數據變化。
也就是說，必須可以在主節點寫了新數據後，及時複製這些變化的數據，所謂及時，不能拉下太多哦.
那麼，符合上面兩個條件的節點就能夠認爲是存活的，也能夠認爲是同步的（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.heartbeatACK(15sec);
    if(!isAlive) {
        node.numNotAlive += 1;
        if(node.numNotAlive >= 3) {
            node.declareDeadOrFailed();
            slave-nodes.remove(node);
            
            //回調也可 leader-node-app.notifyNodeDeadOrFailed(node)
            
        }
    }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();

關於第二點，要稍微複雜點了，怎麼搞呢？
來這麼分析：

1：數據 messages.:
2：操做 op-log.
3：偏移 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) {
        //30秒內沒有回覆，node卡住了
        nodeSucked = true;
    }
    if(node.logOffsetDiff > 100) {
        //node複製跟不上了，差距超過100條數據
        nodeSucked = true;
    }

    if(nodeSucked) {
        //總之node「死」掉了，其實到底死沒死，誰知道呢？network-error在分佈式系統中或者節點失敗這個事情是正常現象.
        node.declareDeadOrFailed();
        //不和你玩啦，集羣不要你了
        nodes.remove(node);
        //該怎麼處理呢，拋個事件吧.
        fire-event-NodeDeadOrFailed(node);
    }
});
});

timer.run();

上面的節點的狀態管理通常由zookeeper來作，leader或者master節點也會維護那麼點狀態。

那麼應用中的leader或者master節點，只須要從zookeeper拉狀態就能夠，同時，上面的實現是否是必定最佳呢？不是的，並且多數操做能夠合起來，但爲了描述節點是否存活這個事兒，我們這麼寫沒啥問題。

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

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

1：同時多個slave節點失敗了？
沒有100%的可用性.數據中心和機房癱瘓、網絡電纜切斷、hacker入侵刪了你的根，總之你rp爆表了.
2：若是主節點失敗了，那master-master不行嘛？
keep-alived或者LVS或者你本身寫failover吧.

高可用架構（HA）又是個大件兒了，此文不展開了。
咱們來關注下recover方面的東西，這裏把視野打開點，不只關注slave節點重啓後追log來同步數據，咱們看下在實際應用中，數據請求（包括讀、寫、更新）失敗怎麼辦？

你們可能都會說，重試（retry）唄、重放（replay）唄或者乾脆無論了唄！
行，都行，這些都是策略，但具體怎麼個搞法，你真的清楚了？

一個bigdata問題

咱們先擺個探討的背景：

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

Reach is the number of unique people exposed to a URL on Twitter.

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

怎麼解決呢？

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

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

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

方法 ：getUrlToTweetersMap(String url_id)

SQL ： /* 數據庫A，表url_user存儲了轉發某url的user */
SELECT url_user.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 users.id as user_id
FROM users
WHERE users.followee_id = ${tweeter_id}

返回：tweeter的粉絲

.

/** ---------------------------------
* 3. 求出Reach
* __________________________________*/

var url = queryArgs.getUrl();
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：數據量有多大？
不知道讀者有木有對這個問題的數據庫I/O有點想法，或者虎軀一震呢？
Computing reach is too intense for a single machine – it can require thousands of database calls and tens of millions of tuples.
在上面的數據庫設計中避免了JOIN，爲了提升求大V粉絲的性能，能夠將一批大V做爲batch/bulk，而後多個batch併發讀，誓死搞死數據庫。
這裏將微博到轉發者表所在的庫，與粉絲庫分離，若是數據更大怎麼辦？
庫再分表...

OK，假設你已經很是熟悉傳統關係型數據庫的分庫分表及數據路由（讀路徑的聚合、寫路徑的分發）、或者你對於sharding技術也很熟悉、或者你良好的結合了HBase的橫向擴展能力並有一致性策略來解決其二級索引問題.
總之，存儲和讀取的問題假設你已經解決了，那麼分佈式計算呢？

2：微博這種應用，人與人之間的關係成圖狀（網），你怎麼建模存儲？而不只僅對應這個問題，好比：
某人的好友的好友可能和某人有幾分相熟？
看看用storm怎麼來解決分佈式計算，並提供流式計算的能力：

// url到大V -> 數據庫1
 TridentState urlToTweeters =
 topology.newStaticState(getUrlToTweetersState());
 // 大V到粉絲 -> 數據庫2
 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）

回到主題，引出上面的例子，一是爲了引出一個有關分佈式（存儲+計算）的問題，二是透漏這麼點意思：
碼農，就應該關注設計和實現的東西，好比Jay Kreps是如何發明Kafka這個輪子的 : ]

若是你仍是碼農級別，咱來務點實吧，前面咱們說到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
}

若是說能夠保證以下三點：

一、事務ID的生成是強有序的.（隔離性，串行）
二、同一個事務ID對應的一批數據相同.（冪等性，屢次操做一個結果）
三、單條數據會且僅會出如今某批數據中.（一致性，無遺漏無重複）

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

// 更新後數據
{
 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的值，以方便分庫、分表後，應用層的數據路由書寫.
這個內容也太多，話題也太大，就不在此展開了。

你如今知道twitter的snowflake生成全局惟一且有序的ID的重要性了。

兩階段提交

如今用zookeeper來作兩階段提交已是入門級技術，因此也不展開了。

若是你的數據庫不支持原子操做，那麼考慮兩階段提交吧。

若是想免費學習Java工程化、高性能及分佈式、深刻淺出。微服務、Spring，MyBatis，Netty源碼分析的朋友能夠加個人Java進階羣：478030634，羣裏有阿里大牛直播講解技術，以及Java大型互聯網技術的視頻免費分享給你們。

結語

To be continued.