GemFire 分布式事務(wù)

1、 2009 VMware Inc. All rights reservedGemFire 分布式事分布式事務(wù)務(wù)GemFire 中國(guó)社區(qū)發(fā)起人 楊旭鈞2分布式系統(tǒng)中的事務(wù)分布式系統(tǒng)中的事務(wù)對(duì)于分布式Region來(lái)說(shuō)，數(shù)據(jù)經(jīng)?？缍鄠€(gè)緩存節(jié)點(diǎn)分布，但是事務(wù)只在一個(gè)緩存節(jié)點(diǎn)管理，因此可能多個(gè)region會(huì)參與到一個(gè)事務(wù)中。事務(wù)操作首先作用在數(shù)據(jù)主拷貝節(jié)點(diǎn)，然后分布到其他成員，而不考慮是哪個(gè)成員先開(kāi)始的緩存操作。-運(yùn)行事務(wù)代碼的成員被稱為事務(wù)初始化器.-管理事務(wù)和數(shù)據(jù)的成員被稱為事務(wù)數(shù)據(jù)節(jié)點(diǎn)事務(wù)數(shù)據(jù)節(jié)點(diǎn)可能是事務(wù)初始化器也可能不是。當(dāng)事務(wù)提交時(shí)，事務(wù)數(shù)據(jù)節(jié)點(diǎn)發(fā)布這份數(shù)據(jù)到其他節(jié)點(diǎn)。分區(qū)區(qū)域的事務(wù)分區(qū)區(qū)域

2、的事務(wù)如下圖所示：M1運(yùn)行著兩個(gè)事務(wù)，T1和T2。T1作用在M1的Y數(shù)據(jù)上，因此M1既是事務(wù)初始化器也是事務(wù)數(shù)據(jù)節(jié)點(diǎn)。T2作用在M2的X和W數(shù)據(jù)上，因此M1是事務(wù)初始化器，M2是事務(wù)數(shù)據(jù)節(jié)點(diǎn)。事務(wù)在數(shù)據(jù)節(jié)點(diǎn)上管理，包括事務(wù)視圖、所有操作、所有本地緩存事件處理，當(dāng)T2被提交時(shí)，M2的緩存被更新，事務(wù)事件跨系統(tǒng)來(lái)分布，好像此事務(wù)源于M2。3分布式系統(tǒng)中的事務(wù)分布式系統(tǒng)中的事務(wù)復(fù)制區(qū)域的事務(wù)復(fù)制區(qū)域的事務(wù)對(duì)于復(fù)制區(qū)域，事務(wù)和其操作被應(yīng)用到本地成員，然后事務(wù)的狀態(tài)被發(fā)布到其他成員，而數(shù)據(jù)如何傳播，則根據(jù)region的屬性設(shè)置。 locallocal非分布式, 在本地處理事務(wù)沖突. 與partition

3、ed region不兼容 distributed-no-ackdistributed-no-ack本地處理事務(wù)沖突, 成員之間缺乏協(xié)調(diào). distributed-ackdistributed-ack即在本地處理事務(wù)沖突，又在成員之間處理事務(wù)沖突4事務(wù)工作原理事務(wù)工作原理Transaction viewTransaction view在并發(fā)訪問(wèn)緩存的時(shí)候，事務(wù)之間是隔離的。每一個(gè)事務(wù)都有自己的私有視圖，包括已讀取的條目和其變更。當(dāng)一個(gè)數(shù)據(jù)條目進(jìn)入事務(wù)的時(shí)候，將在事務(wù)視圖上生成一個(gè)數(shù)據(jù)狀態(tài)的快照，此事務(wù)能夠保存數(shù)據(jù)的原有狀態(tài)，快照的另外一個(gè)用處是在提交階段恢復(fù)寫(xiě)沖突。此事務(wù)維護(hù)著當(dāng)前條目的視圖，用

4、來(lái)在事務(wù)中做出的變更。當(dāng)提交成功時(shí)，在事務(wù)視圖中記錄的變更被合并到緩存上。如果提交失敗或者事務(wù)回滾，所有的變更都將丟棄。當(dāng)事務(wù)提交時(shí)，事務(wù)管理系統(tǒng)使用“two-phase commit protocol”：1. 通過(guò)駐留線程， 預(yù)留事務(wù)中所有的變更條目（對(duì)于分區(qū)區(qū)域，在事務(wù)運(yùn)行的數(shù)據(jù)節(jié)點(diǎn)上做預(yù)留）2. 在受影響的鍵上檢查緩存，確保所有的數(shù)據(jù)條目都處于相同的狀態(tài)。-如果有任何沖突，管理器回滾事務(wù)-如果沒(méi)有沖突，管理器則執(zhí)行如下步驟：調(diào)用TransactionWriter，將事務(wù)更新寫(xiě)到本地緩存，傳播更新到其他成員。在更新的緩存上，調(diào)用CacheListener，在事務(wù)運(yùn)行的成員上調(diào)用Transa

5、ctionListener。3. 在條目上釋放事務(wù)預(yù)留管理器更新本地緩存，以非原子方式發(fā)布更新到其他成員。如果其他線程讀取事務(wù)正在修改的鍵，他們可能是在預(yù)處理或者后處理的狀態(tài)來(lái)查看。如果是其他線程要修改事務(wù)正在修改的鍵，這個(gè)變更可能和事務(wù)更新混合在一起。當(dāng)事務(wù)提交之后，此時(shí)事務(wù)處理的結(jié)果可能和原有的預(yù)期不一致。5distributed-ack scope6distributed-ack scope7事務(wù)事件事務(wù)事件事務(wù)處理過(guò)程中事務(wù)處理過(guò)程中當(dāng)事務(wù)在運(yùn)行時(shí)，每一個(gè)事務(wù)的操作都被傳遞到CacheWriter，如果操作在事務(wù)之外，CacheWriter則丟棄此操作。CacheWriter的每一次操

6、作都被應(yīng)用到緩存上的事務(wù)視圖上，進(jìn)入到TransactionEvent對(duì)象的CacheEvent列表中。事件合并事件合并緩存事件被合并，如果一個(gè)鍵已經(jīng)有了一個(gè)事件，則此事件被移除，同時(shí)當(dāng)前的操作被添加到列表的末尾。如下所示：Region.create (A, W);Region.put (A, valX);Region.put (B, valQ);Region.invalidate (A);Region.put (A, valY);所產(chǎn)生的事件被存儲(chǔ)在CacheEvent列表。事務(wù)提交事務(wù)提交當(dāng)事務(wù)被提交時(shí)，Gemfire傳遞Transaction Event到Transaction Writ

7、er上。在提交之后：在提交之后：Gemfire傳遞Transaction Event到Transaction Listener。遍歷Cache Event列表，為每一個(gè)操作調(diào)用所有的Listener。分布TransactionEvent到所有感興趣的緩存。8事務(wù)的核心類(lèi)事務(wù)的核心類(lèi)com.gemstone.gemfire.cachecom.gemstone.gemfire.cacheCacheTransactionManagerCacheTransactionManager管理緩存中事務(wù)和的事務(wù)應(yīng)用補(bǔ)丁.TransactionEventTransactionEvent在事務(wù)中所做的操作.Tra

8、nsactionIdTransactionId全局唯一的事務(wù)標(biāo)識(shí)符.TransactionWriterTransactionWriter在提交之前調(diào)事務(wù)應(yīng)用. 通常情況下，用于同步后臺(tái)的數(shù)據(jù)源.TransactionListenerTransactionListener在提交之后處理事務(wù)事件的應(yīng)用補(bǔ)丁.9Transaction APICache c = CacheFactory.create(DistributedSystem.connect(null);AttributesFactory af = new AttributesFactory();af.setDataPolicy(DataPo

9、licy.REPLICATE);Region cash = c.createRegion(cash, af.create();Region trades = c.createRegion(trades, af.create();CacheTransactionManager txmgr = c.getCacheTransactionManager();try txmgr.begin();final String customer = Customer1;final Integer purchase = Integer.valueOf(1000);/ Decrement cashInteger cashBalance = cash.get(customer);Integer newBalance =Integer.valueOf(cashBalance != null ? cashBalance : 0)- purchase);cash.put(customer, newBalance);/ Increment tradesInteger tradeBalance = trades.get(customer);newBalance =Integer.val