微吐槽

hello,world.

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

• 本文略長，讀者需要有一定耐心，如果你是高級碼農或者架構師級別，你可以跳過。
• 本文注重實戰或者實現，不涉及CAP，略提ACID。
• 本文適合基礎分布式程序員：

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

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

集群中存活的節點與同步

分布式系統中，如何判斷一個節點(node)是否存活?

kafka這樣認為：

1. 此節點和zookeeper能喊話.(Keep sessions with zookeeper through heartbeats.)
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.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(); 

 

關于第二點，要稍微復雜點了，怎么搞呢?

來這么分析：

• 數據 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) { 
            //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拉狀態就可以，同時，上面的實現是不是一定最佳呢?不是的，而且多數操作可以合起來，但為了描述節點是否存活這個事兒，咱們這么寫沒啥問題。

#p#

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

好嘛，終于說到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啊!

#p#

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

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的方式來保證數據復制和一致性，那么我們會怎么處理一致性問題呢?

主節點有新數據寫入.

從節點追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中，如果這樣追，會導致這批數據重復。 這種場景，從語義上來講，就是數據最少處理一次，意味著數據處理會重復。

#p#

僅處理一次(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相同，按事務要求這次數據應該已經處理過了，跳過?

#p#

不，這種事不能靠猜的，想想我們有的幾個性質，其中關鍵一點就是：

給定一批數據，它們所屬的事務ID相同。 仔細體會下，上面那句話和下面這句話的差別： 給定一個事務ID，任何時候，其所關聯的那批數據相同。

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

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

 

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

 

你發現了什么呢?

不同的事務ID，導致了不同的值：

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

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

這就是Opaque Transaction.

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

Transaction：

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

Opaque-Transaction：

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

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

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

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

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

兩階段提交

現在用zookeeper來做兩階段提交已經是入門級技術，所以也不展開了。

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

To be continued.

博文出處：http://www.cnblogs.com/foreach-break/p/distributed_system_and_transaction.html