亚洲激情中文在线,美女黄网久久,澳门成人av

什么是Paxos共識算法

最初的服務往往都是通過單體架構對外提供的，即單Server-單Database模式。隨著業務的不斷擴展，用戶和請求數都在不斷上升，如何應對大量的請求就成了每個服務都需要解決的問題，這也就是我們常說的高并發。為了解決單臺服務器面對高并發的蒼白無力，可以通過增加服務器數量來解決，即多Server-單Database（Master-Slave）模式，此時的壓力就來到了數據庫一方，數據庫的IO效率決定了整個服務的效率，繼續增加Server數量將無法提升服務性能。這就衍生出了當前火熱的微服務架構。當用戶請求經由負載均衡分配到某一服務實例上后，如何保證該服務的其他實例最終能夠得到相同的數據變化呢？這就要用到Paxos分布式共識協議，Paxos解決的就是共識問題，也就是一段時間后，無論get哪一個服務實例，都能獲取到相同的數據。目前國內外的分布式產品很多都使用了Paxos協議，可以說Paxos幾乎就是共識協議的標準和代名詞。

Paxos有兩種協議，我們常常提到的其實是Basic Paxos，另一種叫Multi Paxos，如無特殊說明，本文中提到的Paxos協議均為Basic Paxos。

Paxos協議是由圖靈獎獲得者Leslie Lamport于1998年在其論文《The Part-Time Parliament》中首次提出的，講述了一個希臘小島Paxos是如何通過決議的。但由于該論文晦澀艱深，當時的計算機界大牛們也沒幾個人能理解。于是Lamport2001年再次發表了《Paxos Made Simple》，摘要部分是這么寫的：

The Paxos algorithm, when presented in plain English, is very simple.

翻譯過來就是：不會吧，不會吧，這么簡單的Paxos算法不會真的有人弄不懂吧？然而事實卻是很多人對Paxos都望而卻步，理解Paxos其實并不難，但是Paxos的難點在于工程化，如何利用Paxos協議寫出一個能過夠真正在生產環境中跑起來的服務才是Paxos最難的地方，關于Paxos的工程化可以參考微信后臺團隊撰寫的《微信自研生產級paxos類庫PhxPaxos實現原理介紹》

Paxos如何保證一致性的

Paxos協議一共有兩個階段：Prepare和Propose，兩種角色：Proposer和Acceptor，每一個服務實例既是Proposer，同時也是Acceptor，Proposer負責提議，Acceptor決定是否接收來自Proposer的提議，一旦提議被多數接受，那么我們就可以宣稱對該提議包含的值達成了一致，而且不會再改變。

階段一：Prepare 準備

Proposer生成全局唯一ProposalID（時間戳+ServerID）
Proposer向所有Acceptor（包括Proposer自己）發送Prepare(n = ProposalID)請求
Acceptor比較n和minProposal, if n > minProposal, minProposal = n，Acceptor返回已接受的提議(acceptedProposal, acceptedValue)
承諾1：不再接受n = minProposal的Prepare請求
承諾2：不再接受n minProposal的Propose請求
應答1：返回此前已接受的提議
當Proposer收到大于半數的返回后
Prepare請求被拒絕，重新生成ProposalID并發送Prepare請求
Prepare請求被接受且有已接受的提議，選擇最大的ProposalID對應的值作為提議的值
Prepare請求被接受且沒有已接受的提議，可選擇任意提議值
階段二：Propose 提議
Proposer向所有Acceptor（包括Proposer自己）發送Accept(n=ProposalID,value=ProposalValue)請求
Acceptor比較n和minProposal, if n >= minProposal, minProposal = n, acceptedValue = value，返回已接受的提議（minProposal，acceptedValue）
當Proposer收到大于半數的返回后
Propose請求被拒絕，重新生成ProposalID并發送Prepare請求
Propose請求被接受，則數據達成一致性

一旦提議被半數以上的服務接受，那么我們就可以宣稱整個服務集群在這一提議上達成了一致。

需要注意的是，在一個服務集群中以上兩個階段是很有可能同時發生的。例如：實例A已完成Prepare階段，并發送了Propose請求。同時實例B開始了Prepare階段，并生成了更大的ProposalID發送Prepare請求，可能導致實例A的Propose請求被拒絕。每個服務實例也是同時在扮演Proposer和Acceptor角色，向其他服務發送請求的同時，可能也在處理別的服務發來的請求。

使用GO語言實現Paxos協議

服務注冊與發現

由于每個服務實例都是在執行相同的代碼，那我們要如何知曉其他服務實例的入口呢（IP和端口號）？方法之一就是寫死在代碼中，或者提供一份配置文件。服務啟動后可以讀取該配置文件。但是這種方法不利于維護，一旦我們需要移除或添加服務則需要在每個機器上重新休息配置文件。

除此之外，我們可以通過一個第三方服務：服務的注冊與發現來注冊并獲知當前集群的總服務實例數，即將本地的配置文件改為線上的配置服務。

服務注冊：Register函數，服務實例啟動后通過調用這個RPC方法將自己注冊在服務管理中

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
}

服務發現：GetServers函數，服務通過調用該RPC方法獲取所有服務實例的信息（IP和端口號）

func (s *Service) GetServers(args *GetServersArgs, reply *GetServersReply) error {
 // return all servers
 reply.ServerInfos = s.Servers
 
 return nil
}

Prepare階段

Proposer，向所有的服務發送Prepare請求，并等待直到半數以上的服務返回結果，這里也可以等待所有服務返回后再處理，但是Paxos協議可以容忍小于半數的服務宕機，因此我們只等待大于N/2個返回即可。當返回的結果有任何一個請求被拒絕，那Proposer即認為這次的請求被拒絕，返回重新生成ProposalID并發送新一輪的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，同樣首先向所有的服務發送Propose請求，并等待知道半數以上的服務返回結果。如果返回的結果有任何一個請求被拒絕，則Proposer認為這次的請求被拒絕，返回重新生成ProposalID并發送新一輪的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
}

運行

運行結果：

這里我一共開啟了3個服務實例，并在每次請求之前加入了隨機的延遲，模擬網絡通信中的延遲，因此每個服務的每個請求并不是同時發出的

動圖一張：