基于python的Paxos算法實(shí)現(xiàn)

更新時(shí)間：2019年07月03日 16:37:35 作者：charles_lun

這篇文章主要介紹了基于python的Paxos算法實(shí)現(xiàn)，理解一個(gè)算法最快，最深刻的做法，我覺著可能是自己手動(dòng)實(shí)現(xiàn)，雖然項(xiàng)目中不用自己實(shí)現(xiàn)，有已經(jīng)封裝好的算法庫(kù)，供我們調(diào)用，我覺著還是有必要自己親自實(shí)踐一下,需要的朋友可以參考下

理解一個(gè)算法最快，最深刻的做法，我覺著可能是自己手動(dòng)實(shí)現(xiàn)，雖然項(xiàng)目中不用自己實(shí)現(xiàn)，有已經(jīng)封裝好的算法庫(kù)，供我們調(diào)用，我覺著還是有必要自己親自實(shí)踐一下。

這里首先說明一下，python這種動(dòng)態(tài)語言，對(duì)不熟悉的人可能看著比較別扭，不像java那樣參數(shù)類型是固定的，所以看著會(huì)有些蛋疼。這里環(huán)境用的是python2.7。

class Message:
  # command
  MSG_ACCEPTOR_AGREE = 0 # 追隨者約定
  MSG_ACCEPTOR_ACCEPT = 1 # 追隨者接受
  MSG_ACCEPTOR_REJECT = 2 # 追隨者拒絕-網(wǎng)絡(luò)不通
  MSG_ACCEPTOR_UNACCEPT = 3 # 追隨者網(wǎng)絡(luò)通-不同意
  MSG_ACCEPT = 4 # 接受
  MSG_PROPOSE = 5 # 提議
  MSG_EXT_PROPOSE = 6 # 額外提議
  MSG_HEARTBEAT = 7 # 心跳，每隔一段時(shí)間同步消息
  def __init__(self, command=None):
    self.command = command
  # 把收到的消息原原路返回，作為應(yīng)答消息
  def copyAsReply(self, message):
    # 提議ID #當(dāng)前的ID #發(fā)給誰 #誰發(fā)的
    self.proposalID, self.instanceID, self.to, self.source = message.proposalID, message.instanceID, message.source, message.to
    self.value = message.value # 發(fā)的信息

然后是利用socket,線程和隊(duì)列實(shí)現(xiàn)的消息處理器：

# 基于socket傳遞消息，封裝網(wǎng)絡(luò)傳遞消息
import threading
import pickle
import socket
import queue
class MessagePump(threading.Thread):
  # 收取消息線程
  class MPHelper(threading.Thread):
    #
    def __init__(self, owner):
      self.owner = owner
      threading.Thread.__init__(self)
    def run(self):
      while not self.owner.abort: # 只要所有者線程沒有結(jié)束,一直接受消息
        try:
          (bytes, addr) = self.owner.socket.recvfrom(2048) # 收取消息
          msg = pickle.loads(bytes) # 讀取二進(jìn)制數(shù)據(jù)轉(zhuǎn)化為消息
          msg.source = addr[1]
          self.owner.queue.put(msg) # 隊(duì)列存入消息
        except Exception as e:
          pass

  def __init__(self, owner, port, timeout=2):
    threading.Thread.__init__(self)
    self.owner = owner
    self.abort = False
    self.timeout = 2
    self.port = port
    self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) # UDP通信
    self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 200000) # 通信參數(shù)
    self.socket.bind(("localhost", port)) # 通信地址，ip，端口
    self.socket.settimeout(timeout) # 超時(shí)設(shè)置
    self.queue = queue.Queue() # 隊(duì)列
    self.helper = MessagePump.MPHelper(self) # 接收消息

  # 運(yùn)行主線程
  def run(self):
    self.helper.start() # 開啟收消息的線程
    while not self.abort:
      message = self.waitForMessage() # 阻塞等待
      self.owner.recvMessage(message) # 收取消息

  # 等待消息
  def waitForMessage(self):
    try:
      msg = self.queue.get(True, 3) # 抓取數(shù)據(jù)，最多等待3s
      return msg
    except:
      return None

  # 發(fā)送消息
  def sendMessage(self, message):
    bytes = pickle.dumps(message) # 轉(zhuǎn)化為二進(jìn)制
    address = ("localhost", message.to) # 地址ip,端口(ip,port)
    self.socket.sendto(bytes, address)
    return True
  #是否停止收取消息
  def doAbort(self):
    self.abort = True

再來一個(gè)消息處理器，模擬消息的傳遞，延遲，丟包，其實(shí)這個(gè)類沒什么卵用，這個(gè)是為模擬測(cè)試準(zhǔn)備的

from MessagePump import MessagePump
import random
class AdversarialMessagePump(MessagePump): # 類的繼承
  # 對(duì)抗消息傳輸，延遲消息并任意順序傳遞,模擬網(wǎng)絡(luò)的延遲，消息傳送并不是順序
  def __init__(self, owner, port, timeout=2):
    MessagePump.__init__(self, owner, port, timeout) # 初始化父類
    self.messages = set() # 集合避免重復(fù)

  def waitForMessage(self):
    try:
      msg = self.queue.get(True, 0.1) # 從隊(duì)列抓取數(shù)據(jù)
      self.messages.add(msg) # 添加消息
    except Exception as e: # 處理異常
      pass
      # print(e)
    if len(self.messages) > 0 and random.random() < 0.95: # Arbitrary!
      msg = random.choice(list(self.messages)) # 隨機(jī)抓取消息發(fā)送
      self.messages.remove(msg) # 刪除消息
    else:
      msg = None
    return msg

再來一個(gè)是記錄類

# InstanceRecord本地記錄類，主要記錄追隨者、領(lǐng)導(dǎo)者最高編號(hào)的協(xié)議
from PaxosLeaderProtocol import PaxosLeaderProtocol
class InstanceRecord:
  def __init__(self):
    self.protocols = {}
    self.highestID = (-1, -1) # (port,count)
    self.value = None

  def addProtocol(self, protocol):
    self.protocols[protocol.proposalID] = protocol
    #
    if protocol.proposalID[1] > self.highestID[1] or (
        protocol.proposalID[1] == self.highestID[1] and protocol.proposalID[0] > self.highestID[0]):
      self.highestID = protocol.proposalID # 取得編號(hào)最大的協(xié)議

  def getProtocol(self, protocolID):
    return self.protocols[protocolID]

  def cleanProtocols(self):
    keys = self.protocols.keys()
    for k in keys:
      protocol = self.protocols[k]
      if protocol.state == PaxosLeaderProtocol.STATE_ACCEPTED:
        print("刪除協(xié)議")
        del self.protocols[k]

下面就是Acceptor的實(shí)現(xiàn)：

# 追隨者
from MessagePump import MessagePump
from Message import Message
from InstanceRecord import InstanceRecord
from PaxosAcceptorProtocol import PaxosAcceptorProtocol
class PaxosAcceptor:
  def __init__(self, port, leaders):
    self.port = port
    self.leaders = leaders
    self.instances = {} # 接口列表
    self.msgPump = MessagePump(self, self.port) # 消息傳遞器
    self.failed = False

  # 開始消息傳送
  def start(self):
    self.msgPump.start()

  # 停止
  def stop(self):
    self.msgPump.doAbort()

  # 失敗
  def fail(self):
    self.failed = True

  def recover(self):
    self.failed = False

  # 發(fā)送消息
  def sendMessage(self, message):
    self.msgPump.sendMessage(message)

  # 收消息,只收取為提議的消息
  def recvMessage(self, message):
    if message == None:
      return
    if self.failed: # 失敗狀態(tài)不收取消息
      return

    if message.command == Message.MSG_PROPOSE: # 判斷消息是否為提議
      if message.instanceID not in self.instances:
        record = InstanceRecord() # 記錄器
        self.instances[message.instanceID] = record
      protocol = PaxosAcceptorProtocol(self) # 創(chuàng)建協(xié)議
      protocol.recvProposal(message) # 收取消息
      self.instances[message.instanceID].addProtocol(protocol)
    else:
      self.instances[message.instanceID].getProtocol(message.proposalID).doTransition(message)

  # 通知客戶端,
  def notifyClient(self, protocol, message):
    if protocol.state == PaxosAcceptorProtocol.STATE_PROPOSAL_ACCEPTED: # 提議被接受，通知
      self.instances[protocol.instanceID].value = message.value # 儲(chǔ)存信息
      print(u"協(xié)議被客戶端接受 %s" % message.value)

  # 獲取最高同意的建議
  def getHighestAgreedProposal(self, instance):
    return self.instances[instance].highestID # （port,count）

  # 獲取接口數(shù)據(jù)
  def getInstanceValue(self, instance):
    return self.instances[instance].value

那再看下AcceptorProtocol的實(shí)現(xiàn)：

from Message import Message
class PaxosAcceptorProtocol(object):
  # State variables
  STATE_UNDEFINED = -1 # 協(xié)議沒有定義的情況0
  STATE_PROPOSAL_RECEIVED = 0 # 收到消息
  STATE_PROPOSAL_REJECTED = 1 # 拒絕鏈接
  STATE_PROPOSAL_AGREED = 2 # 同意鏈接
  STATE_PROPOSAL_ACCEPTED = 3 # 同意請(qǐng)求
  STATE_PROPOSAL_UNACCEPTED = 4 # 拒絕請(qǐng)求

  def __init__(self, client):
    self.client = client
    self.state = PaxosAcceptorProtocol.STATE_UNDEFINED

  # 收取,只處理協(xié)議類型的消息
  def recvProposal(self, message):

    if message.command == Message.MSG_PROPOSE: # 協(xié)議
      self.proposalID = message.proposalID
      self.instanceID = message.instanceID
      (port, count) = self.client.getHighestAgreedProposal(message.instanceID) # 端口，協(xié)議內(nèi)容的最高編號(hào)
      # 檢測(cè)編號(hào)處理消息協(xié)議
      # 判斷協(xié)議是否最高 
      if count < self.proposalID[1] or (count == self.proposalID[1] and port < self.proposalID[0]):
        self.state = PaxosAcceptorProtocol.STATE_PROPOSAL_AGREED # 協(xié)議同意
        print("同意協(xié)議:%s, %s " % (message.instanceID, message.value))
        value = self.client.getInstanceValue(message.instanceID)
        msg = Message(Message.MSG_ACCEPTOR_AGREE) # 同意協(xié)議
        msg.copyAsReply(message)
        msg.value = value
        msg.sequence = (port, count)
        self.client.sendMessage(msg) # 發(fā)送消息
      else: # 不再接受比最高協(xié)議小的提議
        self.state = PaxosAcceptorProtocol.STATE_PROPOSAL_REJECTED
      return self.proposalID
    else:
      # 錯(cuò)誤重試
      pass
  # 過度
  def doTransition(self, message): # 如果當(dāng)前協(xié)議狀態(tài)是接受連接，消息類型是接受
    if self.state == PaxosAcceptorProtocol.STATE_PROPOSAL_AGREED and message.command == Message.MSG_ACCEPT:
      self.state = PaxosAcceptorProtocol.STATE_PROPOSAL_ACCEPTED # 接收協(xié)議
      msg = Message(Message.MSG_ACCEPTOR_ACCEPT) # 創(chuàng)造消息
      msg.copyAsReply(message) # 拷貝并回復(fù)
      for l in self.client.leaders:
        msg.to = l
        self.client.sendMessage(msg) # 給領(lǐng)導(dǎo)發(fā)送消息
      self.notifyClient(message) # 通知自己
      return True
    raise Exception("并非預(yù)期的狀態(tài)和命令")

  # 通知 自己客戶端
  def notifyClient(self, message):
    self.client.notifyClient(self, message)

接著看下Leader和LeaderProtocol實(shí)現(xiàn)：

# 領(lǐng)導(dǎo)者
import threading
import Queue
import time
from Message import Message
from MessagePump import MessagePump
from InstanceRecord import InstanceRecord
from PaxosLeaderProtocol import PaxosLeaderProtocol
class PaxosLeader:
  # 定時(shí)監(jiān)聽
  class HeartbeatListener(threading.Thread):
    def __init__(self, leader):
      self.leader = leader
      self.queue = Queue.Queue() # 消息隊(duì)列
      self.abort = False
      threading.Thread.__init__(self)

    def newHB(self, message):
      self.queue.put(message)

    def doAbort(self):
      self.abort = True

    def run(self): # 讀取消息
      elapsed = 0
      while not self.abort:
        s = time.time()
        try:
          hb = self.queue.get(True, 2)
          # 設(shè)定規(guī)則，誰的端口號(hào)比較高，誰就是領(lǐng)導(dǎo)
          if hb.source > self.leader.port:
            self.leader.setPrimary(False)
        except:
          self.leader.setPrimary(True)

  # 定時(shí)發(fā)送
  class HeartbeatSender(threading.Thread):
    def __init__(self, leader):
      threading.Thread.__init__(self)
      self.leader = leader
      self.abort = False
    def doAbort(self):
      self.abort = True
    def run(self):
      while not self.abort:
        time.sleep(1)
        if self.leader.isPrimary:
          msg = Message(Message.MSG_HEARTBEAT)
          msg.source = self.leader.port
          for leader in self.leader.leaders:
            msg.to = leader
            self.leader.sendMessage(msg)

  def __init__(self, port, leaders=None, acceptors=None):
    self.port = port
    if leaders == None:
      self.leaders = []
    else:
      self.leaders = leaders
    if acceptors == None:
      self.acceptors = []
    else:
      self.acceptors = acceptors
    self.group = self.leaders + self.acceptors # 集合合并
    self.isPrimary = False # 自身是不是領(lǐng)導(dǎo)
    self.proposalCount = 0
    self.msgPump = MessagePump(self, port) # 消息傳送器
    self.instances = {}
    self.hbListener = PaxosLeader.HeartbeatListener(self) # 監(jiān)聽
    self.hbSender = PaxosLeader.HeartbeatSender(self) # 發(fā)送心跳
    self.highestInstance = -1 # 協(xié)議狀態(tài)
    self.stoped = True # 是否正在運(yùn)行
    self.lasttime = time.time() # 最后一次時(shí)間

  def sendMessage(self, message):
    self.msgPump.sendMessage(message)

  def start(self):
    self.hbSender.start()
    self.hbListener.start()
    self.msgPump.start()
    self.stoped = False

  def stop(self):
    self.hbSender.doAbort()
    self.hbListener.doAbort()
    self.msgPump.doAbort()
    self.stoped = True

  def setPrimary(self, primary): # 設(shè)置領(lǐng)導(dǎo)者
    if self.isPrimary != primary:
      # Only print if something's changed
      if primary:
        print(u"我是leader%s" % self.port)
      else:
        print(u"我不是leader%s" % self.port)
    self.isPrimary = primary

  # 獲取所有的領(lǐng)導(dǎo)下面的追隨者
  def getGroup(self):
    return self.group

  def getLeaders(self):
    return self.leaders

  def getAcceptors(self):
    return self.acceptors

  # 必須獲得1/2以上的人支持
  def getQuorumSize(self):
    return (len(self.getAcceptors()) / 2) + 1

  def getInstanceValue(self, instanceID):
    if instanceID in self.instances:
      return self.instances[instanceID].value
    return None

  def getHistory(self): # 歷史記錄
    return [self.getInstanceValue(i) for i in range(1, self.highestInstance + 1)]

  # 抓取同意的數(shù)量
  def getNumAccpted(self):
    return len([v for v in self.getHistory() if v != None])

  # 抓取空白時(shí)間處理下事務(wù)
  def findAndFillGaps(self):
    for i in range(1, self.highestInstance):
      if self.getInstanceValue(i) == None:
        print("填充空白", i)
        self.newProposal(0, i)
    self.lasttime = time.time()

  # 采集無用信息
  def garbageCollect(self):
    for i in self.instances:
      self.instances[i].cleanProtocols()

  # 通知領(lǐng)導(dǎo)
  def recvMessage(self, message):
    if self.stoped:
      return
    if message == None:
      if self.isPrimary and time.time() - self.lasttime > 15.0:
        self.findAndFillGaps()
        self.garbageCollect()
      return
    #處理心跳信息
    if message.command == Message.MSG_HEARTBEAT:
      self.hbListener.newHB(message)
      return True
    #處理額外的提議
    if message.command == Message.MSG_EXT_PROPOSE:
      print("額外的協(xié)議", self.port, self.highestInstance)
      if self.isPrimary:
        self.newProposal(message.value)
      return True

    if self.isPrimary and message.command != Message.MSG_ACCEPTOR_ACCEPT:
      self.instances[message.instanceID].getProtocol(message.proposalID).doTransition(message)

    if message.command == Message.MSG_ACCEPTOR_ACCEPT:
      if message.instanceID not in self.instances:
        self.instances[message.instanceID] = InstanceRecord()
      record = self.instances[message.instanceID]
      if message.proposalID not in record.protocols:#創(chuàng)建協(xié)議
        protocol = PaxosLeaderProtocol(self)
        protocol.state = PaxosLeaderProtocol.STATE_AGREED
        protocol.proposalID = message.proposalID
        protocol.instanceID = message.instanceID
        protocol.value = message.value
        record.addProtocol(protocol)
      else:
        protocol = record.getProtocol(message.proposalID)

      protocol.doTransition(message)

    return True
  # 新建提議
  def newProposal(self, value, instance=None):
    protocol = PaxosLeaderProtocol(self)
    if instance == None: # 創(chuàng)建協(xié)議標(biāo)號(hào)
      self.highestInstance += 1
      instanceID = self.highestInstance
    else:
      instanceID = instance
    self.proposalCount += 1
    id = (self.port, self.proposalCount)
    if instanceID in self.instances:
      record = self.instances[instanceID]
    else:
      record = InstanceRecord()
      self.instances[instanceID] = record
    protocol.propose(value, id, instanceID)
    record.addProtocol(protocol)

  def notifyLeader(self, protocol, message):
    if protocol.state == PaxosLeaderProtocol.STATE_ACCEPTED:
      print("協(xié)議接口%s被%s接受" % (message.instanceID, message.value))
      self.instances[message.instanceID].accepted = True
      self.instances[message.instanceID].value = message.value
      self.highestInstance = max(message.instanceID, self.highestInstance)
      return
    if protocol.state == PaxosLeaderProtocol.STATE_REJECTED: # 重新嘗試
      self.proposalCount = max(self.proposalCount, message.highestPID[1])
      self.newProposal(message.value)
      return True
    if protocol.state == PaxosLeaderProtocol.STATE_UNACCEPTED:
      pass

LeaderProtocol實(shí)現(xiàn)：

from Message import Message
class PaxosLeaderProtocol(object):
  STATE_UNDEFINED = -1 # 協(xié)議沒有定義的情況0
  STATE_PROPOSED = 0 # 協(xié)議消息
  STATE_REJECTED = 1 # 拒絕鏈接
  STATE_AGREED = 2 # 同意鏈接
  STATE_ACCEPTED = 3 # 同意請(qǐng)求
  STATE_UNACCEPTED = 4 # 拒絕請(qǐng)求
  def __init__(self, leader):
    self.leader = leader
    self.state = PaxosLeaderProtocol.STATE_UNDEFINED
    self.proposalID = (-1, -1)
    self.agreecount, self.acceptcount = (0, 0)
    self.rejectcount, self.unacceptcount = (0, 0)
    self.instanceID = -1
    self.highestseen = (0, 0)
  # 提議
  def propose(self, value, pID, instanceID):
    self.proposalID = pID
    self.value = value
    self.instanceID = instanceID
    message = Message(Message.MSG_PROPOSE)
    message.proposalID = pID
    message.instanceID = instanceID
    message.value = value
    for server in self.leader.getAcceptors():
      message.to = server
      self.leader.sendMessage(message)
    self.state = PaxosLeaderProtocol.STATE_PROPOSED

    return self.proposalID

  # 過度
  def doTransition(self, message):
    # 根據(jù)狀態(tài)運(yùn)行協(xié)議
    if self.state == PaxosLeaderProtocol.STATE_PROPOSED:
      if message.command == Message.MSG_ACCEPTOR_AGREE:
        self.agreecount += 1
        if self.agreecount >= self.leader.getQuorumSize(): # 選舉
          print(u"達(dá)成協(xié)議的法定人數(shù)，最后的價(jià)值回答是:%s" % message.value)
          if message.value != None:
            if message.sequence[0] > self.highestseen[0] or (
                message.sequence[0] == self.highestseen[0] and message.sequence[1] > self.highestseen[
              1]):
              self.value = message.value
              self.highestseen = message.sequence

            self.state = PaxosLeaderProtocol.STATE_AGREED # 同意更新
            # 發(fā)送同意消息
            msg = Message(Message.MSG_ACCEPT)
            msg.copyAsReply(message)
            msg.value = self.value
            msg.leaderID = msg.to
            for server in self.leader.getAcceptors():
              msg.to = server
              self.leader.sendMessage(msg)
            self.leader.notifyLeader(self, message)
          return True

        if message.command == Message.MSG_ACCEPTOR_REJECT:
          self.rejectcount += 1
          if self.rejectcount >= self.leader.getQuorumSize():
            self.state = PaxosLeaderProtocol.STATE_REJECTED
            self.leader.notifyLeader(self, message)
          return True

    if self.state == PaxosLeaderProtocol.STATE_AGREED:
      if message.command == Message.MSG_ACCEPTOR_ACCEPT: # 同意協(xié)議
        self.acceptcount += 1
        if self.acceptcount >= self.leader.getQuorumSize():
          self.state = PaxosLeaderProtocol.STATE_ACCEPTED # 接受
          self.leader.notifyLeader(self, message)
      if message.command == Message.MSG_ACCEPTOR_UNACCEPT:
        self.unacceptcount += 1
        if self.unacceptcount >= self.leader.getQuorumSize():
          self.state = PaxosLeaderProtocol.STATE_UNACCEPTED
          self.leader.notifyLeader(self, message)

測(cè)試模塊：

import socket, pickle, time
from Message import Message
from PaxosAcceptor import PaxosAcceptor
from PaxosLeader import PaxosLeader

if __name__ == "__main__":
  # 設(shè)定5個(gè)客戶端
  numclients = 5
  clients = [PaxosAcceptor(port, [54321, 54322]) for port in range(64320, 64320 + numclients)]
  # 兩個(gè)領(lǐng)導(dǎo)者
  leader1 = PaxosLeader(54321, [54322], [c.port for c in clients])
  leader2 = PaxosLeader(54322, [54321], [c.port for c in clients])

  # 開啟領(lǐng)導(dǎo)者與追隨者
  leader1.start()
  leader1.setPrimary(True)
  leader2.setPrimary(True)
  leader2.start()
  for c in clients:
    c.start()

  # 破壞,客戶端不鏈接
  clients[0].fail()
  clients[1].fail()

  # 通信
  s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) # udp協(xié)議
  start = time.time()
  for i in range(1000):
    m = Message(Message.MSG_EXT_PROPOSE) # 消息
    m.value = 0 + i # 消息參數(shù)
    m.to = 54322 # 設(shè)置傳遞的端口
    bytes = pickle.dumps(m) # 提取的二進(jìn)制數(shù)據(jù)
    s.sendto(bytes, ("localhost", m.to)) # 發(fā)送消息

  while leader2.getNumAccpted() < 999:
    print("休眠的這一秒 %d " % leader2.getNumAccpted())
    time.sleep(1)
  print(u"休眠10秒")
  time.sleep(10)
  print(u"停止leaders")
  leader1.stop()
  leader2.stop()
  print(u"停止客戶端")
  for c in clients:
    c.stop()
  print(u"leader1歷史紀(jì)錄")
  print(leader1.getHistory())
  print(u"leader2歷史紀(jì)錄")
  print(leader2.getHistory())
  end = time.time()
  print(u"一共用了%f秒" % (end - start))

代碼確實(shí)比較長(zhǎng)，看起來有些困難，最好還是在pycharm上看這個(gè)邏輯，可以快速定位參數(shù)指向，如果有不對(duì)的地方歡迎指正

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: