你真的很熟分布式和事务吗？
微吐槽
hello,world.
不想了，我等碼農，還是看看怎么來處理分布式系統中的事務這個老大難吧！
本文略長，讀者需要有一定耐心，如果你是高級碼農或者架構師級別，你可以跳過。
本文注重實戰或者實現，不涉及CAP，略提ACID。
本文適合基礎分布式程序員：
由于分布式這個話題太大，事務這個話題也太大，我們從一個集群的一個小小節點開始談起。
集群中存活的節點與同步
分布式系統中，如何判斷一個節點（node）是否存活？
kafka這樣認為：
也就是說，必須能夠在主節點寫了新數據后，及時復制這些變化的數據，所謂及時，不能拉下太多哦.
那么，符合上面兩個條件的節點就可以認為是存活的，也可以認為是同步的（in-sync）.
關于第1點，大家對心跳都很熟悉，那么我們可以這樣認為某個節點不能和zookeeper喊話了：
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 | zookeeper-node: var timer = newtimer() .setInterval(10sec) .onTime(slave-nodes,function(slave-nodes){ ????slave-nodes.forEach( node -> { ????????booleanisAlive = 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 = newtimer() .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 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 | // 1. 先考慮messages // 2. 再考慮log的postion或者offset // 3. 考慮msg和off都記錄在同源數據庫或者存儲設備上.(database or storage-device.) vartimer = newtimer() .setInterval(10sec) .onTime(slave-nodes,function(nodes){ ????varcore-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節點在，不影響數據讀取。
沒有100%的可用性.數據中心和機房癱瘓、網絡電纜切斷、hacker入侵刪了你的根，總之你rp爆表了.
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 2 3 4 5 6 7 8 9 10 11 12 | /** --------------------------------- * 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] |
| 1 2 3 4 5 6 7 8 9 10 11 12 | /** --------------------------------- * 2. 求出大V的粉絲 * __________________________________*/ 方法 : getFollowers(String tweeter_id); SQL :?? /* 數據庫B */ SELECT users.id as user_id FROM users WHERE users.followee_id = ${tweeter_id} 返回：tweeter的粉絲 |
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 | /** --------------------------------- * 3. 求出Reach * __________________________________*/ varurl = queryArgs.getUrl(); vartweeters = getUrlToTweetersMap(); varresult = newHashMap<String,Integer>(); tweeters.forEach(t -> { ????// 你可以批量in + 并發讀來優化下面方法的性能 ????varfollowers = getFollowers(t.tweeter_id); ????followers.forEach(f -> { ????????//hash去重 ????????result.put(f.user_id,1); ????}); }); //Reach returnresult.size(); |
其實這又引出了一個很重要的問題，也是很多大談框架、設計、模式卻往往忽視的問題：性能和數據庫建模的關系。
不知道讀者有木有對這個問題的數據庫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怎么來解決分布式計算，并提供流式計算的能力：
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 | // url到大V -> 數據庫1 TridentState urlToTweeters = ????topology.newStaticState(getUrlToTweetersState()); // 大V到粉絲 -> 數據庫2 TridentState tweetersToFollowers = ????topology.newStaticState(getTweeterToFollowersState()); topology.newDRPCStream("reach") ????.stateQuery(urlToTweeters,newFields("args"),newMapGet(), newFields("tweeters")) ????.each(newFields("tweeters"),newExpandList(), newFields("tweeter")) ????.shuffle()/* 大V的粉絲很多，所以需要分布式處理*/ ????.stateQuery(tweetersToFollowers,newFields("tweeter"),newMapGet(), newFields("followers")) ????.parallelismHint(200)/* 粉絲很多，所以需要高并發 */ ????.each(newFields("followers"),newExpandList(), newFields("follower")) ????.groupBy(newFields("follower")) ????.aggregate(newOne(), newFields("one"))/* 去重 */ ????.parallelismHint(20) ????.aggregate(newCount(), newFields("reach"));/* 計算reach數 */ |
最多處理一次（At most once）
回到主題，引出上面的例子，一是為了引出一個有關分布式（存儲+計算）的問題，二是透漏這么點意思：
碼農，就應該關注設計和實現的東西，比如Jay Kreps是如何發明Kafka這個輪子的 : ]
如果你還是碼農級別，咱來務點實吧，前面我們說到recover，節點恢復的問題，那么我們恢復幾個東西？
基本的：
- 節點狀態
- 節點數據
本篇從數據上來討論下這個問題，為使問題再簡單點，我們考慮寫數據的場景，如果我們用write-ahead-log的方式來保證數據復制和一致性，那么我們會怎么處理一致性問題呢？
(1). 把數據的id偏移寫入log;
(2). 正要處理數據本身，從節點掛了。
那么根據上文的節點存活條件，這個從節點掛了這件事被探測到了，從節點由維護人員手動或者其自己恢復了，那么在加入集群和小伙伴們繼續玩耍之前，它要同步自己的狀態和數據。
問題來了：
如果根據log內的數據偏移來同步數據，那么，因為這個節點在處理數據之前就把偏移寫好了，可是那批數據lost-datas沒有得到處理，如果追log之后的數據來同步，那么那批數據lost-datas就丟了。
在這種情況下，就叫作數據最多處理一次，也就是說數據會丟失。
最少處理一次（At least once）
好吧，丟失數據不能容忍，那么我們換種方式來處理：
(1). 先處理數據；
(2). 正要把數據的id偏移寫入log，從節點掛了。
問題又來了：
如果從節點追log來同步數據，那么因為那批數據duplicated-datas被處理過了，而數據偏移沒有反映到log中，如果這樣追，會導致這批數據重復。
這種場景，從語義上來講，就是數據最少處理一次，意味著數據處理會重復。
僅處理一次（Exactly once）
Transaction
好吧，數據重復也不能容忍？要求挺高啊。
大家都追求的強一致性保證（這里是最終一致性），怎么來搞呢？
換句話說，在更新數據的時候，事務能力如何保障呢？
假設一批數據如下：
| 1 2 3 4 5 6 | // 新到數據 { ????transactionId:4 ????urlId:99 ????reach:5 } |
現在要更新這批數據到庫里或者log里，那么原來的情況是：
| 1 2 3 4 5 6 | // 老數據 { ????transactionId：3 ????urlId:99 ????reach:3 } |
如果說可以保證如下三點：
那么，放心大膽的更新好了：
| 1 2 3 4 5 6 7 | // 更新后數據 { ????transactionId：4 ????urlId:99 ????//3 + 5 = 8 ????reach:8 } |
注意到這個更新是ID偏移和數據一起更新的，那么這個操作靠什么來保證：原子性。
你的數據庫不提供原子性？后文略有提及。
這里是更新成功了。如果更新的時候，節點掛了，那么庫里或者log里的id偏移不寫，數據也不處理，等節點恢復，就可以放心去同步，然后加入集群玩耍了。
所以說，要保證數據僅處理一次，還是挺困難的吧？
上面的保障“僅處理一次”這個語義的實現有什么問題呢？
性能問題。
這里已經使用了batch策略來減少到庫或磁盤的Round-Trip Time，那么這里的性能問題是什么呢？
考慮一下，采用master-master架構來保證主節點的可用性，但是一個主節點失敗了，到另一個主節點主持工作，是需要時間的。
假設從節點正在同步，啪！主節點掛了！因為要保證僅處理一次的語義，所以原子性發揮作用，失敗，回滾，然后從主節點拉失敗的數據（你不能就近更新，因為這批數據可能已經變化了，或者你根本沒緩存本批數據），結果是什么呢？
老主節點掛了， 新的主節點還沒啟動，所以這次事務就卡在這里，直到數據同步的源——主節點可以響應請求。
如果不考慮性能，就此作罷，這也不是什么大事。
你似乎意猶未盡？來吧，看看“銀彈”是什么？
Opaque-Transaction
現在，我們來追求這樣一種效果：
某條數據在一批數據中（這批數據對應著一個事務），很可能會失敗，但是它會在另一批數據中成功。
換句話說，一批數據的事務ID一定相同。
來看看例子吧，老數據不變，只是多了個字段：prevReach。
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 | // 老數據 { ????transactionId：3 ????urlId:99 ????//注意這里多了個字段，表示之前的reach的值 ????prevReach:2 ????reach:3 } // 新到數據 { ????transactionId:4 ????urlId:99 ????reach:5 } |
這種情況，新事務的ID更大、更靠后，表明新事務可以執行，還等什么，直接更新，更新后數據如下：
| 1 2 3 4 5 6 7 8 9 | // 新到數據 { ????transactionId:4 ????urlId:99 ????//注意這里更新為之前的值 ????prevReach:3 ????//3 + 5 = 8 ????reach:8 } |
現在來看下另外的情況：
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 | // 老數據 { ????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的值來更新：
| 1 2 3 4 5 6 7 8 9 | // 更新后數據 { ????transactionId:3 ????urlId:99 ????//這個值不變 ????prevReach:2 ????//2 + 5 = 7 ????reach:7 } |
你發現了什么呢？
不同的事務ID，導致了不同的值：
這就是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.
from:?http://www.importnew.com/23597.html
總結
以上是生活随笔為你收集整理的你真的很熟分布式和事务吗？的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 浅谈Spring框架注解的用法分析
- 下一篇： 多线程之线程池的应用