微吐槽##
hello,world.
不想了，我等碼農，還是看看怎么來處理分布式系統中的事務這個老大難吧！
本文略長，讀者需要有一定耐心，如果你是高級碼農或者架構師級別，你可以跳過。
本文注重實戰或者實現，不涉及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();
關於第二點，要稍微復雜點了，怎么搞呢？
來這么分析：
- 數據 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拉狀態就可以，同時，上面的實現是不是一定最佳呢？不是的，而且多數操作可以合起來，但為了描述節點是否存活這個事兒，咱們這么寫沒啥問題。
節點死掉、失敗、不同步了，咋處理呢？
好嘛，終於說到failover和recover了，那failover比較簡單，因為還有其它的slave節點在，不影響數據讀取。
- 同時多個slave節點失敗了？
沒有100%的可用性.數據中心和機房癱瘓、網絡電纜切斷、hacker入侵刪了你的根，總之你rp爆表了. - 如果主節點失敗了，那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啊！
其實這又引出了一個很重要的問題，也是很多大談框架、設計、模式卻往往忽視的問題：性能和數據庫建模的關系。
- 數據量有多大？
不知道讀者有木有對這個問題的數據庫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的橫向擴展能力並有一致性策略來解決其二級索引問題.
總之，存儲和讀取的問題假設你已經解決了，那么分布式計算呢？ - 微博這種應用，人與人之間的關系成圖狀（網），你怎么建模存儲？而不僅僅對應這個問題，比如：
某人的好友的好友可能和某人有幾分相熟？
看看用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）##
好吧，丟失數據不能容忍，那么我們換種方式來處理：
- 主節點有新數據寫入.
- 從節點追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，導致了不同的值：
- 當事務ID為4，大於存儲中的事務ID3，Reach更新為3+5 = 8.
- 當事務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來做兩階段提交已經是入門級技術，所以也不展開了。
如果你的數據庫不支持原子操作，那么考慮兩階段提交吧。
結語##
To be continued.