什么是Paxos共識算法
最初的服務(wù)往往都是通過單體架構(gòu)對外提供的,即單Server-單Database模式。隨著業(yè)務(wù)的不斷擴展,用戶和請求數(shù)都在不斷上升,如何應(yīng)對大量的請求就成了每個服務(wù)都需要解決的問題,這也就是我們常說的高并發(fā)。為了解決單臺服務(wù)器面對高并發(fā)的蒼白無力,可以通過增加服務(wù)器數(shù)量來解決,即多Server-單Database(Master-Slave)模式,此時的壓力就來到了數(shù)據(jù)庫一方,數(shù)據(jù)庫的IO效率決定了整個服務(wù)的效率,繼續(xù)增加Server數(shù)量將無法提升服務(wù)性能。這就衍生出了當(dāng)前火熱的微服務(wù)架構(gòu)。當(dāng)用戶請求經(jīng)由負載均衡分配到某一服務(wù)實例上后,如何保證該服務(wù)的其他實例最終能夠得到相同的數(shù)據(jù)變化呢?這就要用到Paxos分布式共識協(xié)議,Paxos解決的就是共識問題,也就是一段時間后,無論get哪一個服務(wù)實例,都能獲取到相同的數(shù)據(jù)。目前國內(nèi)外的分布式產(chǎn)品很多都使用了Paxos協(xié)議,可以說Paxos幾乎就是共識協(xié)議的標準和代名詞。
Paxos有兩種協(xié)議,我們常常提到的其實是Basic Paxos,另一種叫Multi Paxos,如無特殊說明,本文中提到的Paxos協(xié)議均為Basic Paxos。
Paxos協(xié)議是由圖靈獎獲得者Leslie Lamport于1998年在其論文《The Part-Time Parliament》中首次提出的,講述了一個希臘小島Paxos是如何通過決議的。但由于該論文晦澀艱深,當(dāng)時的計算機界大牛們也沒幾個人能理解。于是Lamport2001年再次發(fā)表了《Paxos Made Simple》,摘要部分是這么寫的:
The Paxos algorithm, when presented in plain English, is very simple.
翻譯過來就是:不會吧,不會吧,這么簡單的Paxos算法不會真的有人弄不懂吧?然而事實卻是很多人對Paxos都望而卻步,理解Paxos其實并不難,但是Paxos的難點在于工程化,如何利用Paxos協(xié)議寫出一個能過夠真正在生產(chǎn)環(huán)境中跑起來的服務(wù)才是Paxos最難的地方,關(guān)于Paxos的工程化可以參考微信后臺團隊撰寫的《微信自研生產(chǎn)級paxos類庫PhxPaxos實現(xiàn)原理介紹》
Paxos如何保證一致性的
Paxos協(xié)議一共有兩個階段:Prepare和Propose,兩種角色:Proposer和Acceptor,每一個服務(wù)實例既是Proposer,同時也是Acceptor,Proposer負責(zé)提議,Acceptor決定是否接收來自Proposer的提議,一旦提議被多數(shù)接受,那么我們就可以宣稱對該提議包含的值達成了一致,而且不會再改變。
階段一:Prepare 準備
- Proposer生成全局唯一ProposalID(時間戳+ServerID)
- Proposer向所有Acceptor(包括Proposer自己)發(fā)送Prepare(n = ProposalID)請求
- Acceptor比較n和minProposal, if n > minProposal, minProposal = n,Acceptor返回已接受的提議(acceptedProposal, acceptedValue)
- 承諾1:不再接受n = minProposal的Prepare請求
- 承諾2:不再接受n minProposal的Propose請求
- 應(yīng)答1:返回此前已接受的提議
- 當(dāng)Proposer收到大于半數(shù)的返回后
- Prepare請求被拒絕,重新生成ProposalID并發(fā)送Prepare請求
- Prepare請求被接受且有已接受的提議,選擇最大的ProposalID對應(yīng)的值作為提議的值
- Prepare請求被接受且沒有已接受的提議,可選擇任意提議值
階段二:Propose 提議
- Proposer向所有Acceptor(包括Proposer自己)發(fā)送Accept(n=ProposalID,value=ProposalValue)請求
- Acceptor比較n和minProposal, if n >= minProposal, minProposal = n, acceptedValue = value,返回已接受的提議(minProposal,acceptedValue)
- 當(dāng)Proposer收到大于半數(shù)的返回后
- Propose請求被拒絕,重新生成ProposalID并發(fā)送Prepare請求
- Propose請求被接受,則數(shù)據(jù)達成一致性
一旦提議被半數(shù)以上的服務(wù)接受,那么我們就可以宣稱整個服務(wù)集群在這一提議上達成了一致。
需要注意的是,在一個服務(wù)集群中以上兩個階段是很有可能同時發(fā)生的。 例如:實例A已完成Prepare階段,并發(fā)送了Propose請求。同時實例B開始了Prepare階段,并生成了更大的ProposalID發(fā)送Prepare請求,可能導(dǎo)致實例A的Propose請求被拒絕。 每個服務(wù)實例也是同時在扮演Proposer和Acceptor角色,向其他服務(wù)發(fā)送請求的同時,可能也在處理別的服務(wù)發(fā)來的請求。
使用GO語言實現(xiàn)Paxos協(xié)議
服務(wù)注冊與發(fā)現(xiàn)
由于每個服務(wù)實例都是在執(zhí)行相同的代碼,那我們要如何知曉其他服務(wù)實例的入口呢(IP和端口號)?方法之一就是寫死在代碼中,或者提供一份配置文件。服務(wù)啟動后可以讀取該配置文件。但是這種方法不利于維護,一旦我們需要移除或添加服務(wù)則需要在每個機器上重新休息配置文件。
除此之外,我們可以通過一個第三方服務(wù):服務(wù)的注冊與發(fā)現(xiàn)來注冊并獲知當(dāng)前集群的總服務(wù)實例數(shù),即將本地的配置文件改為線上的配置服務(wù)。
服務(wù)注冊:Register函數(shù),服務(wù)實例啟動后通過調(diào)用這個RPC方法將自己注冊在服務(wù)管理中
func (s *Service) Register(args *RegisterArgs, reply *RegisterReply) error {
s.mu.Lock()
defer s.mu.Unlock()
server := args.ServerInfo
for _, server := range s.Servers {
if server.IPAddress == args.ServerInfo.IPAddress server.Port == args.ServerInfo.Port {
reply.Succeed = false
return nil
}
}
reply.ServerID = len(s.Servers)
reply.Succeed = true
s.Servers = append(s.Servers, server)
fmt.Printf("Current registerd servers:\n%v\n", s.Servers)
return nil
}
服務(wù)發(fā)現(xiàn):GetServers函數(shù),服務(wù)通過調(diào)用該RPC方法獲取所有服務(wù)實例的信息(IP和端口號)
func (s *Service) GetServers(args *GetServersArgs, reply *GetServersReply) error {
// return all servers
reply.ServerInfos = s.Servers
return nil
}
Prepare階段
Proposer,向所有的服務(wù)發(fā)送Prepare請求,并等待直到半數(shù)以上的服務(wù)返回結(jié)果,這里也可以等待所有服務(wù)返回后再處理,但是Paxos協(xié)議可以容忍小于半數(shù)的服務(wù)宕機,因此我們只等待大于N/2個返回即可。當(dāng)返回的結(jié)果有任何一個請求被拒絕,那Proposer即認為這次的請求被拒絕,返回重新生成ProposalID并發(fā)送新一輪的Prepare請求。
func (s *Server) CallPrepare(allServers []ServerInfo, proposal Proposal) PrepareReply {
returnedReplies := make([]PrepareReply, 0)
for _, otherS := range allServers {
// use a go routine to call every server
go func(otherS ServerInfo) {
delay := rand.Intn(10)
time.Sleep(time.Second * time.Duration(delay))
args := PrepareArgs{s.Info, proposal.ID}
reply := PrepareReply{}
fmt.Printf("【Prepare】Call Prepare on %v:%v with proposal id %v\n", otherS.IPAddress, otherS.Port, args.ProposalID)
if Call(otherS, "Server.Prepare", args, reply) {
if reply.HasAcceptedProposal {
fmt.Printf("【Prepare】%v:%v returns accepted proposal: %v\n", otherS.IPAddress, otherS.Port, reply.AcceptedProposal)
} else {
fmt.Printf("【Prepare】%v:%v returns empty proposal\n", otherS.IPAddress, otherS.Port)
}
s.mu.Lock()
returnedReplies = append(returnedReplies, reply)
s.mu.Unlock()
}
}(otherS)
}
for {
// wait for responses from majority
if len(returnedReplies) > (len(allServers))/2.0 {
checkReplies := returnedReplies
// three possible response
// 1. deny the prepare, and return an empty/accepted proposal
// as the proposal id is not higher than minProposalID on server (proposal id = server.minProposalID)
// 2. accept the prepare, and return an empty proposal as the server has not accept any proposal yet
// 3. accept the prepare, and return an accepted proposal
// check responses from majority
// find the response with max proposal id
acceptedProposal := NewProposal()
for _, r := range checkReplies {
// if any response refused the prepare, this server should resend prepare
if !r.PrepareAccepted {
return r
}
if r.HasAcceptedProposal r.AcceptedProposal.ID > acceptedProposal.ID {
acceptedProposal = r.AcceptedProposal
}
}
// if some other server has accepted proposal, return that proposal with max proposal id
// if no other server has accepted proposal, return an empty proposal
return PrepareReply{HasAcceptedProposal: !acceptedProposal.IsEmpty(), AcceptedProposal: acceptedProposal, PrepareAccepted: true}
}
//fmt.Printf("Waiting for response from majority...\n")
time.Sleep(time.Second * 1)
}
}
Acceptor,通過比較ProposalID和minProposal,如果ProposalID小于等于minProposal,則拒絕該Prepare請求,否則更新minProposal為ProposalID。最后返回已接受的提議
func (s *Server) Prepare(args *PrepareArgs, reply *PrepareReply) error {
s.mu.Lock()
defer s.mu.Unlock()
// 2 promises and 1 response
// Promise 1
// do not accept prepare request which ProposalID = minProposalID
// Promise 2
// do not accept propose request which ProposalID minProposalID
// Response 1
// respond with accepted proposal if any
if reply.PrepareAccepted = args.ProposalID > s.minProposalID; reply.PrepareAccepted {
// ready to accept the proposal with Id s.minProposalID
s.minProposalID = args.ProposalID
}
reply.HasAcceptedProposal = s.readAcceptedProposal()
reply.AcceptedProposal = s.Proposal
return nil
}
Propose階段
Proposer,同樣首先向所有的服務(wù)發(fā)送Propose請求,并等待知道半數(shù)以上的服務(wù)返回結(jié)果。如果返回的結(jié)果有任何一個請求被拒絕,則Proposer認為這次的請求被拒絕,返回重新生成ProposalID并發(fā)送新一輪的Prepare請求
func (s *Server) CallPropose(allServers []ServerInfo, proposal Proposal) ProposeReply {
returnedReplies := make([]ProposeReply, 0)
for _, otherS := range allServers {
go func(otherS ServerInfo) {
delay := rand.Intn(5000)
time.Sleep(time.Millisecond * time.Duration(delay))
args := ProposeArgs{otherS, proposal}
reply := ProposeReply{}
fmt.Printf("【Propose】Call Propose on %v:%v with proposal: %v\n", otherS.IPAddress, otherS.Port, args.Proposal)
if Call(otherS, "Server.Propose", args, reply) {
fmt.Printf("【Propose】%v:%v returns: %v\n", otherS.IPAddress, otherS.Port, reply)
s.mu.Lock()
returnedReplies = append(returnedReplies, reply)
s.mu.Unlock()
}
}(otherS)
}
for {
// wait for responses from majority
if len(returnedReplies) > (len(allServers))/2.0 {
checkReplies := returnedReplies
for _, r := range checkReplies {
if !r.ProposeAccepted {
return r
}
}
return checkReplies[0]
}
time.Sleep(time.Second * 1)
}
}
Acceptor,通過比較ProposalID和minProposal,如果ProposalID小于minProposal,則拒絕該Propose請求,否則更新minProposal為ProposalID,并將提議持久化到本地磁盤中。
func (s *Server) Propose(args *ProposeArgs, reply *ProposeReply) error {
if s.minProposalID = args.Proposal.ID {
s.mu.Lock()
s.minProposalID = args.Proposal.ID
s.Proposal = args.Proposal
s.SaveAcceptedProposal()
s.mu.Unlock()
reply.ProposeAccepted = true
}
reply.ProposalID = s.minProposalID
return nil
}
運行
運行結(jié)果:
這里我一共開啟了3個服務(wù)實例,并在每次請求之前加入了隨機的延遲,模擬網(wǎng)絡(luò)通信中的延遲,因此每個服務(wù)的每個請求并不是同時發(fā)出的
動圖一張:

靜態(tài)結(jié)果一張:

可以看到3個服務(wù)盡管一開始會嘗試以他們自己的端口號(5001,5002,5003)作為提議值,在Prepare/Propose失敗后,都會重新生成更大的ProposalID并開啟新一輪的提議過程(Prepare,Propose),且最后都以5003達成一致。
小結(jié)
至此,我們就用GO實現(xiàn)了Paxos協(xié)議的核心邏輯。但顯而易見的是,這段代碼仍然存在很多問題,完全無法滿足生產(chǎn)環(huán)境的需求
- 通過channel而不是mutex鎖來共享數(shù)據(jù)
- 如何處理服務(wù)實例的移除和增加
- 如何避免陷入活鎖
到此這篇關(guān)于使用GO實現(xiàn)Paxos共識算法的文章就介紹到這了,更多相關(guān)GO實現(xiàn)Paxos共識算法內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
您可能感興趣的文章:- Golang實現(xiàn)拓撲排序(DFS算法版)
- 圖解Golang的GC垃圾回收算法
- golang實現(xiàn)分頁算法實例代碼
- C++實現(xiàn)分水嶺算法(Watershed Algorithm)