當(dāng)前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

分布式一致性协议Raft原理与实例

發(fā)布時(shí)間：2024/9/20 编程问答 21 豆豆

生活随笔收集整理的這篇文章主要介紹了分布式一致性协议Raft原理与实例小編覺得挺不錯(cuò)的,現(xiàn)在分享給大家,幫大家做個(gè)參考.

來源:http://m635674608.iteye.com/blog/2283621

1.Raft協(xié)議

1.1 Raft簡介

Raft是由Stanford提出的一種更易理解的一致性算法，意在取代目前廣為使用的Paxos算法。目前，在各種主流語言中都有了一些開源實(shí)現(xiàn)，比如本文中將使用的基于JGroups的Raft協(xié)議實(shí)現(xiàn)。關(guān)于Raft的原理，強(qiáng)烈推薦動(dòng)畫版Raft講解。

1.2 Raft原理

在Raft中，每個(gè)結(jié)點(diǎn)會(huì)處于下面三種狀態(tài)中的一種：

follower：所有結(jié)點(diǎn)都以follower的狀態(tài)開始。如果沒收到leader消息則會(huì)變成candidate狀態(tài)
candidate：會(huì)向其他結(jié)點(diǎn)“拉選票”，如果得到大部分的票則成為leader。這個(gè)過程就叫做Leader選舉(Leader Election)
leader：所有對(duì)系統(tǒng)的修改都會(huì)先經(jīng)過leader。每個(gè)修改都會(huì)寫一條日志(log entry)。leader收到修改請求后的過程如下，這個(gè)過程叫做日志復(fù)制(Log Replication)：?
復(fù)制日志到所有follower結(jié)點(diǎn)(replicate entry)
大部分結(jié)點(diǎn)響應(yīng)時(shí)才提交日志
通知所有follower結(jié)點(diǎn)日志已提交
所有follower也提交日志
現(xiàn)在整個(gè)系統(tǒng)處于一致的狀態(tài)

1.2.1 Leader Election

當(dāng)follower在選舉超時(shí)時(shí)間(election timeout)內(nèi)未收到leader的心跳消息(append entries)，則變成candidate狀態(tài)。為了避免選舉沖突，這個(gè)超時(shí)時(shí)間是一個(gè)150~300ms之間的隨機(jī)數(shù)。

成為candidate的結(jié)點(diǎn)發(fā)起新的選舉期(election term)去“拉選票”：

重置自己的計(jì)時(shí)器

投自己一票

發(fā)送?Request Vote消息

如果接收結(jié)點(diǎn)在新term內(nèi)沒有投過票那它就會(huì)投給此candidate，并重置它自己的選舉超時(shí)時(shí)間。candidate拉到大部分選票就會(huì)成為leader，并定時(shí)發(fā)送心跳——Append Entries消息，去重置各個(gè)follower的計(jì)時(shí)器。當(dāng)前Term會(huì)繼續(xù)直到某個(gè)follower接收不到心跳并成為candidate。

如果不巧兩個(gè)結(jié)點(diǎn)同時(shí)成為candidate都去“拉票”怎么辦？這時(shí)會(huì)發(fā)生Splite Vote情況。兩個(gè)結(jié)點(diǎn)可能都拉到了同樣多的選票，難分勝負(fù)，選舉失敗，本term沒有l(wèi)eader。之后又有計(jì)時(shí)器超時(shí)的follower會(huì)變成candidate，將term加一并開始新一輪的投票。

1.2.2 Log Replication

當(dāng)發(fā)生改變時(shí)，leader會(huì)復(fù)制日志給follower結(jié)點(diǎn)，這也是通過Append Entries心跳消息完成的。前面已經(jīng)列舉了Log Replication的過程，這里就不重復(fù)了。

Raft能夠正確地處理網(wǎng)絡(luò)分區(qū)（“腦裂”）問題。假設(shè)A~E五個(gè)結(jié)點(diǎn)，B是leader。如果發(fā)生“腦裂”，A、B成為一個(gè)子分區(qū)，C、D、E成為一個(gè)子分區(qū)。此時(shí)C、D、E會(huì)發(fā)生選舉，選出C作為新term的leader。這樣我們在兩個(gè)子分區(qū)內(nèi)就有了不同term的兩個(gè)leader。這時(shí)如果有客戶端寫A時(shí)，因?yàn)锽無法復(fù)制日志到大部分follower所以日志處于uncommitted未提交狀態(tài)。而同時(shí)另一個(gè)客戶端對(duì)C的寫操作卻能夠正確完成，因?yàn)镃是新的leader，它只知道D和E。

當(dāng)網(wǎng)絡(luò)通信恢復(fù)，B能夠發(fā)送心跳給C、D、E了，卻發(fā)現(xiàn)“改朝換代”了，因?yàn)镃的term值更大，所以B自動(dòng)降格為follower。然后A和B都回滾未提交的日志，并從新leader那里復(fù)制最新的日志。但這樣是不是就會(huì)丟失更新？

2.JGroups-raft介紹

2.1 JGroups中的Raft

JGroups是Java里比較流行的網(wǎng)絡(luò)通信框架，近期順應(yīng)潮流，它也推出了Raft基于JGroups的實(shí)現(xiàn)。簡單試用了一下，還比較容易上手，底層Raft的內(nèi)部機(jī)制都被API屏蔽掉了。下面就通過一個(gè)分布式計(jì)數(shù)器的實(shí)例來學(xué)習(xí)一下Raft協(xié)議在JGroups中的實(shí)際用法。

Maven依賴如下：

Prettyprint代碼??

<code?class="language-xml?hljs??has-numbering">????<span?class="hljs-tag"><<span?class="hljs-title">dependency>??

????????<span?class="hljs-tag"><<span?class="hljs-title">groupId>org.jgroups<span?class="hljs-tag"></<span?class="hljs-title">groupId>??

????????<span?class="hljs-tag"><<span?class="hljs-title">artifactId>jgroups-raft<span?class="hljs-tag"></<span?class="hljs-title">artifactId>??

????????<span?class="hljs-tag"><<span?class="hljs-title">version>0.2<span?class="hljs-tag"></<span?class="hljs-title">version>??

????<span?class="hljs-tag"></<span?class="hljs-title">dependency></code>??

其實(shí)JGroups-raft的Jar包中已經(jīng)自帶了一個(gè)Counter的Demo，但仔細(xì)看了一下，有的地方寫的有些麻煩，不太容易把握住Raft這根主線。所以這里就參照官方的例子，進(jìn)行了簡寫，突出Raft協(xié)議的基本使用方法。JGroups-raft目前資料不多，InfoQ上的這篇文章很不錯(cuò)，還有官方文檔。

2.2 核心API

使用JGroups-raft時(shí)，我們一般會(huì)實(shí)現(xiàn)兩個(gè)接口：RAFT.RoleChange和StateMachine：

實(shí)現(xiàn)RAFT.RoleChange接口的方法能通知我們當(dāng)前哪個(gè)結(jié)點(diǎn)是leader
實(shí)現(xiàn)StateMachine執(zhí)行要實(shí)現(xiàn)一致性的操作

典型單點(diǎn)服務(wù)實(shí)現(xiàn)方式就是：

Prettyprint代碼??

<code?class="language-java?hljs??has-numbering">JChannel?ch?=?<span?class="hljs-keyword">null;??

RaftHandle?handle?=?<span?class="hljs-keyword">new?RaftHandle(ch,?<span?class="hljs-keyword">this);??

handle.addRoleListener(role?->?{??

????<span?class="hljs-keyword">if(role?==?Role.Leader)??

????????<span?class="hljs-comment">//?start?singleton?services??

????<span?class="hljs-keyword">else??

????????<span?class="hljs-comment">//?stop?singleton?services??

});</code>??

2.3 默認(rèn)配置

jgroups-raft.jar中已經(jīng)帶了一個(gè)raft.xml配置文件，作為實(shí)例程序我們可以直接使用它。

簡要解釋一下最核心的幾個(gè)配置項(xiàng)，參照GitHub上的文檔：

UDP：IP多播配置
raft.NO_DUPES：是否檢測新加入結(jié)點(diǎn)的ID與老結(jié)點(diǎn)有重復(fù)
raft.ELECTION：選舉超時(shí)時(shí)間的隨機(jī)化范圍
raft.RAFT：所有Raft集群的成員必須在這里聲明，也可以在運(yùn)行時(shí)通過addServer/removeServer動(dòng)態(tài)修改
raft.REDIRECT：是否轉(zhuǎn)發(fā)請求給leader
raft.CLIENT：在哪個(gè)IP和端口上接收客戶端請求

Prettyprint代碼??

<code?class="language-xml?hljs??has-numbering"><span?class="hljs-comment"><!--??

??Default?stack?using?IP?multicasting.?It?is?similar?to?the?"udp"??

??stack?in?stacks.xml,?but?doesn't?use?streaming?state?transfer?and?flushing??

??author:?Bela?Ban??

-->??

<span?class="hljs-tag"><<span?class="hljs-title">config?<span?class="hljs-attribute">xmlns=<span?class="hljs-value">"urn:org:jgroups"??

????????<span?class="hljs-attribute">xmlns:xsi=<span?class="hljs-value">"http://www.w3.org/2001/XMLSchema-instance"??

????????<span?class="hljs-attribute">xsi:schemaLocation=<span?class="hljs-value">"urn:org:jgroups?http://www.jgroups.org/schema/jgroups.xsd">??

????<span?class="hljs-tag"><<span?class="hljs-title">UDP??

?????????<span?class="hljs-attribute">mcast_addr=<span?class="hljs-value">"228.5.5.5"??

?????????<span?class="hljs-attribute">mcast_port=<span?class="hljs-value">"${jgroups.udp.mcast_port:45588}"??

?????????<span?class="hljs-attribute">...?/>??

????...??

????<span?class="hljs-tag"><<span?class="hljs-title">raft.NO_DUPES/>??

????<span?class="hljs-tag"><<span?class="hljs-title">raft.ELECTION?<span?class="hljs-attribute">election_min_interval=<span?class="hljs-value">"100"?<span?class="hljs-attribute">election_max_interval=<span?class="hljs-value">"500"/>??

????<span?class="hljs-tag"><<span?class="hljs-title">raft.RAFT?<span?class="hljs-attribute">members=<span?class="hljs-value">"A,B,C"?<span?class="hljs-attribute">raft_id=<span?class="hljs-value">"${raft_id:undefined}"/>??

????<span?class="hljs-tag"><<span?class="hljs-title">raft.REDIRECT/>??

????<span?class="hljs-tag"><<span?class="hljs-title">raft.CLIENT?<span?class="hljs-attribute">bind_addr=<span?class="hljs-value">"0.0.0.0"?/>??

<span?class="hljs-tag"></<span?class="hljs-title">config></code>??

3.JGroups-raft實(shí)例

實(shí)例很簡單，只有JGroupsRaftTest和CounterService兩個(gè)類組成。JGroupsRaftTest是測試啟動(dòng)類，而CounterService就是利用Raft協(xié)議實(shí)現(xiàn)的分布式計(jì)數(shù)服務(wù)類。

3.1 JGroupsRaftTest

JGroupsRaftTest的職責(zé)主要有三個(gè)：

創(chuàng)建Raft協(xié)議的JChannel
創(chuàng)建CounterService
循環(huán)讀取用戶輸入

目前簡單實(shí)現(xiàn)了幾種操作包括：初始化計(jì)數(shù)器、加一、減一、讀取計(jì)數(shù)器、查看Raft日志、做Raft快照（用于壓縮日志文件）等。其中對(duì)計(jì)數(shù)器的操作，因?yàn)橐c其他Raft成員進(jìn)行分布式通信，所以當(dāng)前集群必須要多于一個(gè)結(jié)點(diǎn)時(shí)才能進(jìn)行操作。如果要支持單結(jié)點(diǎn)時(shí)的操作，需要做特殊處理。

Prettyprint代碼??

<code?class="language-java?hljs??has-numbering"><span?class="hljs-keyword">import?org.jgroups.JChannel;??

<span?class="hljs-keyword">import?org.jgroups.protocols.raft.RAFT;??

<span?class="hljs-keyword">import?org.jgroups.util.Util;??

<span?class="hljs-javadoc">/**??

?*?Test?jgroups?raft?algorithm?implementation.??

?*/??

<span?class="hljs-keyword">public?<span?class="hljs-class"><span?class="hljs-keyword">class?<span?class="hljs-title">JGroupsRaftTest?{??

????<span?class="hljs-keyword">private?<span?class="hljs-keyword">static?<span?class="hljs-keyword">final?String?CLUSTER_NAME?=?<span?class="hljs-string">"ctr-cluster";??

????<span?class="hljs-keyword">private?<span?class="hljs-keyword">static?<span?class="hljs-keyword">final?String?COUNTER_NAME?=?<span?class="hljs-string">"counter";??

????<span?class="hljs-keyword">private?<span?class="hljs-keyword">static?<span?class="hljs-keyword">final?String?RAFT_XML?=?<span?class="hljs-string">"raft.xml";??

????<span?class="hljs-keyword">public?<span?class="hljs-keyword">static?<span?class="hljs-keyword">void?<span?class="hljs-title">main(String[]?args)?<span?class="hljs-keyword">throws?Exception?{??

????????JChannel?ch?=?<span?class="hljs-keyword">new?JChannel(RAFT_XML).name(args[<span?class="hljs-number">0]);??

????????CounterService?counter?=?<span?class="hljs-keyword">new?CounterService(ch);??

????????<span?class="hljs-keyword">try?{??

????????????doConnect(ch,?CLUSTER_NAME);??

????????????doLoop(ch,?counter);??

????????}?<span?class="hljs-keyword">finally?{??

????????????Util.close(ch);??

????????}??

????}??

????<span?class="hljs-keyword">private?<span?class="hljs-keyword">static?<span?class="hljs-keyword">void?<span?class="hljs-title">doConnect(JChannel?ch,?String?clusterName)?<span?class="hljs-keyword">throws?Exception?{??

????????ch.connect(clusterName);??

????}??

????<span?class="hljs-keyword">private?<span?class="hljs-keyword">static?<span?class="hljs-keyword">void?<span?class="hljs-title">doLoop(JChannel?ch,?CounterService?counter)?{??

????????<span?class="hljs-keyword">boolean?looping?=?<span?class="hljs-keyword">true;??

????????<span?class="hljs-keyword">while?(looping)?{??

????????????<span?class="hljs-keyword">int?key?=?Util.keyPress(<span?class="hljs-string">"\n[0]?Create?[1]?Increment?[2]?Decrement?[3]?Dump?log?[4]?Snapshot?[x]?Exit\n"?+??

????????????????????<span?class="hljs-string">"first-applied="?+?((RAFT)?ch.getProtocolStack().findProtocol(RAFT.class)).log().firstApplied()?+??

????????????????????<span?class="hljs-string">",?last-applied="?+?counter.lastApplied()?+??

????????????????????<span?class="hljs-string">",?commit-index="?+?counter.commitIndex()?+??

????????????????????<span?class="hljs-string">",?log?size="?+?Util.printBytes(counter.logSize())?+?<span?class="hljs-string">":?");??

????????????<span?class="hljs-keyword">if?((key?==?<span?class="hljs-string">'0'?||?key?==?<span?class="hljs-string">'1'?||?key?==?<span?class="hljs-string">'2')?&&?!counter.isLeaderExist())?{??

????????????????System.out.println(<span?class="hljs-string">"Cannot?perform?cause?there?is?no?leader?by?now");??

????????????????<span?class="hljs-keyword">continue;??

????????????}??

????????????<span?class="hljs-keyword">long?val;??

????????????<span?class="hljs-keyword">switch?(key)?{??

????????????????<span?class="hljs-keyword">case?<span?class="hljs-string">'0':??

????????????????????counter.getOrCreateCounter(COUNTER_NAME,?<span?class="hljs-number">1L);??

????????????????????<span?class="hljs-keyword">break;??

????????????????<span?class="hljs-keyword">case?<span?class="hljs-string">'1':??

????????????????????val?=?counter.incrementAndGet(COUNTER_NAME);??

????????????????????System.out.printf(<span?class="hljs-string">"%s:?%s\n",?COUNTER_NAME,?val);??

????????????????????<span?class="hljs-keyword">break;??

????????????????<span?class="hljs-keyword">case?<span?class="hljs-string">'2':??

????????????????????val?=?counter.decrementAndGet(COUNTER_NAME);??

????????????????????System.out.printf(<span?class="hljs-string">"%s:?%s\n",?COUNTER_NAME,?val);??

????????????????????<span?class="hljs-keyword">break;??

????????????????<span?class="hljs-keyword">case?<span?class="hljs-string">'3':??

????????????????????counter.dumpLog();??

????????????????????<span?class="hljs-keyword">break;??

????????????????<span?class="hljs-keyword">case?<span?class="hljs-string">'4':??

????????????????????counter.snapshot();??

????????????????????<span?class="hljs-keyword">break;??

????????????????<span?class="hljs-keyword">case?<span?class="hljs-string">'x':??

????????????????????looping?=?<span?class="hljs-keyword">false;??

????????????????????<span?class="hljs-keyword">break;??

????????????????<span?class="hljs-keyword">case?<span?class="hljs-string">'\n':??

????????????????????System.out.println(COUNTER_NAME?+?<span?class="hljs-string">":?"?+?counter.get(COUNTER_NAME)?+?<span?class="hljs-string">"\n");??

????????????????????<span?class="hljs-keyword">break;??

????????????}??

????????}??

????}??

}</code>??

3.2 CounterService

CounterService是我們的核心類，利用Raft實(shí)現(xiàn)了分布式的計(jì)數(shù)器操作，它的API主要由四部分組成：

Raft Local API：操作本地Raft的狀態(tài)，像日志大小、做快照等
Raft API：實(shí)現(xiàn)Raft的監(jiān)聽器和狀態(tài)機(jī)的方法?
- roleChanged：本地Raft的角色發(fā)生變化
- apply：分布式通信消息
- readContentFrom/writeContentTo：讀寫快照
Counter API：計(jì)數(shù)器的分布式API
Counter Native API：計(jì)數(shù)器的本地API。直接使用的話相當(dāng)于臟讀

Prettyprint代碼??

<code?class="language-java?hljs??has-numbering"><span?class="hljs-keyword">import?org.jgroups.Channel;??

<span?class="hljs-keyword">import?org.jgroups.protocols.raft.RAFT;??

<span?class="hljs-keyword">import?org.jgroups.protocols.raft.Role;??

<span?class="hljs-keyword">import?org.jgroups.protocols.raft.StateMachine;??

<span?class="hljs-keyword">import?org.jgroups.raft.RaftHandle;??

<span?class="hljs-keyword">import?org.jgroups.util.AsciiString;??

<span?class="hljs-keyword">import?org.jgroups.util.Bits;??

<span?class="hljs-keyword">import?org.jgroups.util.ByteArrayDataInputStream;??

<span?class="hljs-keyword">import?org.jgroups.util.ByteArrayDataOutputStream;??

<span?class="hljs-keyword">import?org.jgroups.util.Util;??

<span?class="hljs-keyword">import?java.io.DataInput;??

<span?class="hljs-keyword">import?java.io.DataOutput;??

<span?class="hljs-keyword">import?java.io.IOException;??

<span?class="hljs-keyword">import?java.text.SimpleDateFormat;??

<span?class="hljs-keyword">import?java.util.Date;??

<span?class="hljs-keyword">import?java.util.HashMap;??

<span?class="hljs-keyword">import?java.util.Map;??

<span?class="hljs-javadoc">/**??

?*?Distribute?counter?service?based?on?Raft?consensus?algorithm.??

?*/??

class?CounterService?implements?StateMachine,?RAFT.RoleChange?{??

????<span?class="hljs-keyword">private?RaftHandle?raft;??

????<span?class="hljs-keyword">private?<span?class="hljs-keyword">final?Map<String,?Long>?counters;??

????<span?class="hljs-keyword">private?<span?class="hljs-keyword">enum?Command?{??

????????CREATE,?INCREMENT_AND_GET,?DECREMENT_AND_GET,?GET,?SET??

????}??

????<span?class="hljs-keyword">public?<span?class="hljs-title">CounterService(Channel?ch)?{??

????????<span?class="hljs-keyword">this.raft?=?<span?class="hljs-keyword">new?RaftHandle(ch,?<span?class="hljs-keyword">this);??

????????<span?class="hljs-keyword">this.counters?=?<span?class="hljs-keyword">new?HashMap<>();??

????????raft.raftId(ch.getName())??

????????????.addRoleListener(<span?class="hljs-keyword">this);??

????}??

????<span?class="hljs-comment">//?===========================================??

????<span?class="hljs-comment">//??????????????Raft?Status?API??

????<span?class="hljs-comment">//?===========================================??

????<span?class="hljs-keyword">public?<span?class="hljs-keyword">int?<span?class="hljs-title">lastApplied()?{??

????????<span?class="hljs-keyword">return?raft.lastApplied();??

????}??

????<span?class="hljs-keyword">public?<span?class="hljs-keyword">int?<span?class="hljs-title">commitIndex()?{??

????????<span?class="hljs-keyword">return?raft.commitIndex();??

????}??

????<span?class="hljs-keyword">public?<span?class="hljs-keyword">int?<span?class="hljs-title">logSize()?{??

????????<span?class="hljs-keyword">return?raft.logSize();??

????}??

????<span?class="hljs-keyword">public?<span?class="hljs-keyword">void?<span?class="hljs-title">dumpLog()?{??

????????System.out.println(<span?class="hljs-string">"\nindex?(term):?command\n---------------------");??

????????raft.logEntries((entry,?index)?->?{??

????????????StringBuilder?log?=?<span?class="hljs-keyword">new?StringBuilder()??

????????????????????.append(index)??

????????????????????.append(<span?class="hljs-string">"?(").append(entry.term()).append(<span?class="hljs-string">"):?");??

????????????<span?class="hljs-keyword">if?(entry.command()?==?<span?class="hljs-keyword">null?)?{??

????????????????System.out.println(log.append(<span?class="hljs-string">"<marker?record>"));??

????????????????<span?class="hljs-keyword">return;??

????????????}?<span?class="hljs-keyword">else?<span?class="hljs-keyword">if?(entry.internal())?{??

????????????????System.out.println(log.append(<span?class="hljs-string">"<internal?command>"));??

????????????????<span?class="hljs-keyword">return;??

????????????}??

????????????ByteArrayDataInputStream?in?=?<span?class="hljs-keyword">new?ByteArrayDataInputStream(??

????????????????????entry.command(),?entry.offset(),?entry.length()??

????????????);??

????????????<span?class="hljs-keyword">try?{??

????????????????Command?cmd?=?Command.values()[in.readByte()];??

????????????????String?name?=?Bits.readAsciiString(in).toString();??

????????????????<span?class="hljs-keyword">switch?(cmd)?{??

????????????????????<span?class="hljs-keyword">case?CREATE:??

????????????????????????log.append(cmd)??

????????????????????????????.append(<span?class="hljs-string">"(").append(name).append(<span?class="hljs-string">",?")??

????????????????????????????.append(Bits.readLong(in))??

????????????????????????????.append(<span?class="hljs-string">")");??

????????????????????????<span?class="hljs-keyword">break;??

????????????????????<span?class="hljs-keyword">case?GET:??

????????????????????<span?class="hljs-keyword">case?INCREMENT_AND_GET:??

????????????????????<span?class="hljs-keyword">case?DECREMENT_AND_GET:??

????????????????????????log.append(cmd)??

????????????????????????????.append(<span?class="hljs-string">"(").append(name).append(<span?class="hljs-string">")");??

????????????????????????<span?class="hljs-keyword">break;??

????????????????????<span?class="hljs-keyword">default:??

????????????????????????<span?class="hljs-keyword">throw?<span?class="hljs-keyword">new?IllegalArgumentException(<span?class="hljs-string">"Command?"?+?cmd?+?<span?class="hljs-string">"is?unknown");??

????????????????}??

????????????????System.out.println(log);??

????????????}??

????????????<span?class="hljs-keyword">catch?(IOException?e)?{??

????????????????<span?class="hljs-keyword">throw?<span?class="hljs-keyword">new?IllegalStateException(<span?class="hljs-string">"Error?when?dump?log",?e);??

????????????}??

????????});??

????????System.out.println();??

????}??

????<span?class="hljs-keyword">public?<span?class="hljs-keyword">void?<span?class="hljs-title">snapshot()?{??

????????<span?class="hljs-keyword">try?{??

????????????raft.snapshot();??

????????}?<span?class="hljs-keyword">catch?(Exception?e)?{??

????????????<span?class="hljs-keyword">throw?<span?class="hljs-keyword">new?IllegalStateException(<span?class="hljs-string">"Error?when?snapshot",?e);??

????????}??

????}??

????<span?class="hljs-keyword">public?<span?class="hljs-keyword">boolean?<span?class="hljs-title">isLeaderExist()?{??

????????<span?class="hljs-keyword">return?raft.leader()?!=?<span?class="hljs-keyword">null;??

????}??

????<span?class="hljs-comment">//?===========================================??

????<span?class="hljs-comment">//??????????????Raft?API??

????<span?class="hljs-comment">//?===========================================??

????<span?class="hljs-annotation">@Override??

????<span?class="hljs-keyword">public?<span?class="hljs-keyword">void?<span?class="hljs-title">roleChanged(Role?role)?{??

????????System.out.println(<span?class="hljs-string">"roleChanged?to:?"?+?role);??

????}??

????<span?class="hljs-annotation">@Override??

????<span?class="hljs-keyword">public?<span?class="hljs-keyword">byte[]?<span?class="hljs-title">apply(<span?class="hljs-keyword">byte[]?data,?<span?class="hljs-keyword">int?offset,?<span?class="hljs-keyword">int?length)?<span?class="hljs-keyword">throws?Exception?{??

????????ByteArrayDataInputStream?in?=?<span?class="hljs-keyword">new?ByteArrayDataInputStream(data,?offset,?length);??

????????Command?cmd?=?Command.values()[in.readByte()];??

????????String?name?=?Bits.readAsciiString(in).toString();??

????????System.out.println(<span?class="hljs-string">"["?+?<span?class="hljs-keyword">new?SimpleDateFormat(<span?class="hljs-string">"HH:mm:ss.SSS").format(<span?class="hljs-keyword">new?Date())??

????????????????+?<span?class="hljs-string">"]?Apply:?cmd=["?+?cmd?+?<span?class="hljs-string">"]");??

????????<span?class="hljs-keyword">long?v1,?retVal;??

????????<span?class="hljs-keyword">switch?(cmd)?{??

????????????<span?class="hljs-keyword">case?CREATE:??

????????????????v1?=?Bits.readLong(in);??

????????????????retVal?=?create0(name,?v1);??

????????????????<span?class="hljs-keyword">return?Util.objectToByteBuffer(retVal);??

????????????<span?class="hljs-keyword">case?GET:??

????????????????retVal?=?get0(name);??

????????????????<span?class="hljs-keyword">return?Util.objectToByteBuffer(retVal);??

????????????<span?class="hljs-keyword">case?INCREMENT_AND_GET:??

????????????????retVal?=?add0(name,?<span?class="hljs-number">1L);??

????????????????<span?class="hljs-keyword">return?Util.objectToByteBuffer(retVal);??

????????????<span?class="hljs-keyword">case?DECREMENT_AND_GET:??

????????????????retVal?=?add0(name,?-<span?class="hljs-number">1L);??

????????????????<span?class="hljs-keyword">return?Util.objectToByteBuffer(retVal);??

????????????<span?class="hljs-keyword">default:??

????????????????<span?class="hljs-keyword">throw?<span?class="hljs-keyword">new?IllegalArgumentException(<span?class="hljs-string">"Command?"?+?cmd?+?<span?class="hljs-string">"is?unknown");??

????????}??

????}??

????<span?class="hljs-annotation">@Override??

????<span?class="hljs-keyword">public?<span?class="hljs-keyword">void?<span?class="hljs-title">readContentFrom(DataInput?in)?<span?class="hljs-keyword">throws?Exception?{??

????????<span?class="hljs-keyword">int?size?=?in.readInt();??

????????System.out.println(<span?class="hljs-string">"ReadContentFrom:?size=["?+?size?+?<span?class="hljs-string">"]");??

????????<span?class="hljs-keyword">for?(<span?class="hljs-keyword">int?i?=?<span?class="hljs-number">0;?i?<?size;?i++)?{??

????????????AsciiString?name?=?Bits.readAsciiString(in);??

????????????Long?value?=?Bits.readLong(in);??

????????????counters.put(name.toString(),?value);??

????????}??

????}??

????<span?class="hljs-annotation">@Override??

????<span?class="hljs-keyword">public?<span?class="hljs-keyword">void?<span?class="hljs-title">writeContentTo(DataOutput?out)?<span?class="hljs-keyword">throws?Exception?{??

????????<span?class="hljs-keyword">synchronized?(counters)?{??

????????????<span?class="hljs-keyword">int?size?=?counters.size();??

????????????System.out.println(<span?class="hljs-string">"WriteContentFrom:?size=["?+?size?+?<span?class="hljs-string">"]");??

????????????out.writeInt(size);??

????????????<span?class="hljs-keyword">for?(Map.Entry<String,?Long>?entry?:?counters.entrySet())?{??

????????????????AsciiString?name?=?<span?class="hljs-keyword">new?AsciiString(entry.getKey());??

????????????????Long?value?=?entry.getValue();??

????????????????Bits.writeAsciiString(name,?out);??

????????????????Bits.writeLong(value,?out);??

????????????}??

????????}??

????}??

????<span?class="hljs-comment">//?===========================================??

????<span?class="hljs-comment">//??????????????Counter?API??

????<span?class="hljs-comment">//?===========================================??

????<span?class="hljs-keyword">public?<span?class="hljs-keyword">void?<span?class="hljs-title">getOrCreateCounter(String?name,?<span?class="hljs-keyword">long?initVal)?{??

????????Object?retVal?=?invoke(Command.CREATE,?name,?<span?class="hljs-keyword">false,?initVal);??

????????counters.put(name,?(Long)?retVal);??

????}??

????<span?class="hljs-keyword">public?<span?class="hljs-keyword">long?<span?class="hljs-title">incrementAndGet(String?name)?{??

????????<span?class="hljs-keyword">return?(<span?class="hljs-keyword">long)?invoke(Command.INCREMENT_AND_GET,?name,?<span?class="hljs-keyword">false);??

????}??

????<span?class="hljs-keyword">public?<span?class="hljs-keyword">long?<span?class="hljs-title">decrementAndGet(String?name)?{??

????????<span?class="hljs-keyword">return?(<span?class="hljs-keyword">long)?invoke(Command.DECREMENT_AND_GET,?name,?<span?class="hljs-keyword">false);??

????}??

????<span?class="hljs-keyword">public?<span?class="hljs-keyword">long?<span?class="hljs-title">get(String?name)?{??

????????<span?class="hljs-keyword">return?(<span?class="hljs-keyword">long)?invoke(Command.GET,?name,?<span?class="hljs-keyword">false);??

????}??

????<span?class="hljs-keyword">private?Object?<span?class="hljs-title">invoke(Command?cmd,?String?name,?<span?class="hljs-keyword">boolean?ignoreRetVal,?<span?class="hljs-keyword">long...?values)?{??

????????ByteArrayDataOutputStream?out?=?<span?class="hljs-keyword">new?ByteArrayDataOutputStream(<span?class="hljs-number">256);??

????????<span?class="hljs-keyword">try?{??

????????????out.writeByte(cmd.ordinal());??

????????????Bits.writeAsciiString(<span?class="hljs-keyword">new?AsciiString(name),?out);??

????????????<span?class="hljs-keyword">for?(<span?class="hljs-keyword">long?val?:?values)?{??

????????????????Bits.writeLong(val,?out);??

????????????}??

????????????<span?class="hljs-keyword">byte[]?rsp?=?raft.set(out.buffer(),?<span?class="hljs-number">0,?out.position());??

????????????<span?class="hljs-keyword">return?ignoreRetVal???<span?class="hljs-keyword">null?:?Util.objectFromByteBuffer(rsp);??

????????}??

????????<span?class="hljs-keyword">catch?(IOException?ex)?{??

????????????<span?class="hljs-keyword">throw?<span?class="hljs-keyword">new?RuntimeException(<span?class="hljs-string">"Serialization?failure?(cmd="??

????????????????????+?cmd?+?<span?class="hljs-string">",?name="?+?name?+?<span?class="hljs-string">")",?ex);??

????????}??

????????<span?class="hljs-keyword">catch?(Exception?ex)?{??

????????????<span?class="hljs-keyword">throw?<span?class="hljs-keyword">new?RuntimeException(<span?class="hljs-string">"Raft?set?failure?(cmd="??

????????????????????+?cmd?+?<span?class="hljs-string">",?name="?+?name?+?<span?class="hljs-string">")",?ex);??

????????}??

????}??

????<span?class="hljs-comment">//?===========================================??

????<span?class="hljs-comment">//??????????????Counter?Native?API??

????<span?class="hljs-comment">//?===========================================??

????<span?class="hljs-keyword">public?<span?class="hljs-keyword">synchronized?Long?<span?class="hljs-title">create0(String?name,?<span?class="hljs-keyword">long?initVal)?{??

????????counters.putIfAbsent(name,?initVal);??

????????<span?class="hljs-keyword">return?counters.get(name);??

????}??

????<span?class="hljs-keyword">public?<span?class="hljs-keyword">synchronized?Long?<span?class="hljs-title">get0(String?name)?{??

????????<span?class="hljs-keyword">return?counters.getOrDefault(name,?<span?class="hljs-number">0L);??

????}??

????<span?class="hljs-keyword">public?<span?class="hljs-keyword">synchronized?Long?<span?class="hljs-title">add0(String?name,?<span?class="hljs-keyword">long?delta)?{??

????????Long?oldVal?=?counters.getOrDefault(name,?<span?class="hljs-number">0L);??

????????<span?class="hljs-keyword">return?counters.put(name,?oldVal?+?delta);??

????}??

}</code>??

3.3 運(yùn)行測試

我們分別以A、B、C為參數(shù)，啟動(dòng)三個(gè)JGroupsRaftTest服務(wù)。這樣會(huì)自動(dòng)在C:\Users\cdai\AppData\Local\Temp下生成A.log、B.log、C.log三個(gè)日志文件夾。

Prettyprint代碼??

<code?class="hljs?livecodeserver?has-numbering">cdai@vm?/cygdrive/c/Users/cdai/AppData/Local/Temp??

$?tree?A.<span?class="hljs-built_in">log/?B.<span?class="hljs-built_in">log/?C.<span?class="hljs-built_in">log/??

A.<span?class="hljs-built_in">log/??

|<span?class="hljs-comment">--?000005.sst??

|<span?class="hljs-comment">--?000006.log??

|<span?class="hljs-comment">--?CURRENT??

|<span?class="hljs-comment">--?LOCK??

|<span?class="hljs-comment">--?LOG??

|<span?class="hljs-comment">--?LOG.old??

`<span?class="hljs-comment">--?MANIFEST-000004??

B.<span?class="hljs-built_in">log/??

|<span?class="hljs-comment">--?000003.log??

|<span?class="hljs-comment">--?CURRENT??

|<span?class="hljs-comment">--?LOCK??

|<span?class="hljs-comment">--?LOG??

`<span?class="hljs-comment">--?MANIFEST-000002??

C.<span?class="hljs-built_in">log/??

|<span?class="hljs-comment">--?000003.log??

|<span?class="hljs-comment">--?CURRENT??

|<span?class="hljs-comment">--?LOCK??

|<span?class="hljs-comment">--?LOG??

`<span?class="hljs-comment">--?MANIFEST-000002??

<span?class="hljs-number">0?<span?class="hljs-built_in">directories,?<span?class="hljs-number">17?<span?class="hljs-built_in">files</code>??

3.3.1 分布式一致性

首先A創(chuàng)建計(jì)數(shù)器，B“加一”，C“減一”。可以看到盡管我們是分別在A、B、C上執(zhí)行這三個(gè)操作，但三個(gè)結(jié)點(diǎn)都先后（leader提交日志后通知follower）通過apply()方法收到消息，并在本地的計(jì)數(shù)器Map上同步執(zhí)行操作，保證了數(shù)據(jù)的一致性。最后停掉A服務(wù)，可以看到B通過roleChanged()得到消息，提升為新的Leader，并與C一同繼續(xù)提供服務(wù)。

A的控制臺(tái)輸出：

Prettyprint代碼??

<code?class="hljs?asciidoc?has-numbering"><span?class="hljs-code">-------------------------------------------------------------------??

GMS:?address=A,?cluster=ctr-cluster,?physical?address=2001:0:9d38:6abd:cbb:1f78:3f57:50f6:50100??

-------------------------------------------------------------------??

[0]?Create?[1]?Increment?[2]?Decrement?[3]?Dump?log?[4]?Snapshot?[x]?Exit??

first-applied=0,?last-applied=0,?commit-index=0,?log?size=0b:???

roleChanged?to:?Candidate??

roleChanged?to:?Leader??

0??

<span?class="hljs-attribute">[14:16:00.744]?Apply:?cmd=[CREATE]??

[0]?Create?[1]?Increment?[2]?Decrement?[3]?Dump?log?[4]?Snapshot?[x]?Exit??

first-applied=0,?last-applied=1,?commit-index=1,?log?size=1b:???

<span?class="hljs-attribute">[14:16:07.002]?Apply:?cmd=[INCREMENT_AND_GET]??

<span?class="hljs-attribute">[14:16:14.264]?Apply:?cmd=[DECREMENT_AND_GET]??

3??

<span?class="hljs-header">index?(term):?command??

---------------------??

1?(29):?CREATE(counter,?1)??

2?(29):?INCREMENT<span?class="hljs-emphasis">_AND_GET(counter)??

3?(29):?DECREMENT<span?class="hljs-emphasis">_AND_GET(counter)</code>??

B的控制臺(tái)輸出：

Prettyprint代碼??

<code?class="hljs?sql?has-numbering"><span?class="hljs-comment">-------------------------------------------------------------------??

GMS:?address=B,?cluster=ctr-cluster,?physical?address=2001:0:9d38:6abd:cbb:1f78:3f57:50f6:50101??

<span?class="hljs-comment">-------------------------------------------------------------------??

[0]?<span?class="hljs-operator"><span?class="hljs-keyword">Create?[<span?class="hljs-number">1]?Increment?[<span?class="hljs-number">2]?Decrement?[<span?class="hljs-number">3]?Dump?log?[<span?class="hljs-number">4]?Snapshot?[x]?Exit??

<span?class="hljs-keyword">first-applied=<span?class="hljs-number">0,?<span?class="hljs-keyword">last-applied=<span?class="hljs-number">0,?<span?class="hljs-keyword">commit-index=<span?class="hljs-number">0,?log?<span?class="hljs-keyword">size=<span?class="hljs-number">0b:???

[<span?class="hljs-number">14:<span?class="hljs-number">16:<span?class="hljs-number">01.300]?Apply:?cmd=[<span?class="hljs-keyword">CREATE]??

<span?class="hljs-number">1??

counter:?<span?class="hljs-number">2??

[<span?class="hljs-number">0]?<span?class="hljs-keyword">Create?[<span?class="hljs-number">1]?Increment?[<span?class="hljs-number">2]?Decrement?[<span?class="hljs-number">3]?Dump?log?[<span?class="hljs-number">4]?Snapshot?[x]?Exit??

<span?class="hljs-keyword">first-applied=<span?class="hljs-number">0,?<span?class="hljs-keyword">last-applied=<span?class="hljs-number">2,?<span?class="hljs-keyword">commit-index=<span?class="hljs-number">1,?log?<span?class="hljs-keyword">size=<span?class="hljs-number">2b:???

[<span?class="hljs-number">14:<span?class="hljs-number">16:<span?class="hljs-number">07.299]?Apply:?cmd=[INCREMENT_AND_GET]??

[<span?class="hljs-number">14:<span?class="hljs-number">16:<span?class="hljs-number">14.304]?Apply:?cmd=[DECREMENT_AND_GET]??

roleChanged?<span?class="hljs-keyword">to:?Candidate??

roleChanged?<span?class="hljs-keyword">to:?Leader</code>??

C的控制臺(tái)輸出：

Prettyprint代碼??

<code?class="hljs?sql?has-numbering"><span?class="hljs-comment">-------------------------------------------------------------------??

GMS:?address=C,?cluster=ctr-cluster,?physical?address=2001:0:9d38:6abd:cbb:1f78:3f57:50f6:55800??

<span?class="hljs-comment">-------------------------------------------------------------------??

[<span?class="hljs-number">14:<span?class="hljs-number">16:<span?class="hljs-number">01.300]?Apply:?cmd=[<span?class="hljs-keyword">CREATE]??

[<span?class="hljs-number">14:<span?class="hljs-number">16:<span?class="hljs-number">07.299]?Apply:?cmd=[INCREMENT_AND_GET]??

<span?class="hljs-number">2??

counter:?<span?class="hljs-number">3??

<span?class="hljs-keyword">first-applied=<span?class="hljs-number">0,?<span?class="hljs-keyword">last-applied=<span?class="hljs-number">3,?<span?class="hljs-keyword">commit-index=<span?class="hljs-number">2,?log?<span?class="hljs-keyword">size=<span?class="hljs-number">3b:???

[<span?class="hljs-number">14:<span?class="hljs-number">16:<span?class="hljs-number">14.304]?Apply:?cmd=[DECREMENT_AND_GET]</code>??

3.3.2 服務(wù)恢復(fù)

在只有B和C的集群中，我們執(zhí)行了一次“加一”。當(dāng)我們重新啟動(dòng)A服務(wù)時(shí)，它會(huì)自動(dòng)執(zhí)行這條日志，保持與B和C的一致。從日志的index能夠看出，69是一個(gè)Term，也就是A為Leader時(shí)的“任期”，而70也就是B為Leader時(shí)。

A的控制臺(tái)輸出：

Prettyprint代碼??

<code?class="hljs?asciidoc?has-numbering"><span?class="hljs-code">-------------------------------------------------------------------??

GMS:?address=A,?cluster=ctr-cluster,?physical?address=2001:0:9d38:6abd:cbb:1f78:3f57:50f6:53237??

-------------------------------------------------------------------??

[0]?Create?[1]?Increment?[2]?Decrement?[3]?Dump?log?[4]?Snapshot?[x]?Exit??

first-applied=0,?last-applied=3,?commit-index=3,?log?size=3b:???

<span?class="hljs-attribute">[14:18:45.275]?Apply:?cmd=[INCREMENT_AND_GET]??

<span?class="hljs-attribute">[14:18:45.277]?Apply:?cmd=[GET]??

3??

<span?class="hljs-header">index?(term):?command??

---------------------??

1?(69):?CREATE(counter,?1)??

2?(69):?INCREMENT<span?class="hljs-emphasis">_AND_GET(counter)??

3?(69):?DECREMENT<span?class="hljs-emphasis">_AND_GET(counter)??

4?(70):?INCREMENT<span?class="hljs-emphasis">_AND_GET(counter)??

5?(70):?GET(counter)</code>??

http://blog.csdn.net/dc_726/article/details/48832405

總結(jié)

以上是生活随笔為你收集整理的分布式一致性协议Raft原理与实例的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。