基于python的Paxos算法实现

class Message:
  # command
  MSG_ACCEPTOR_AGREE = 0 # 追随者约定
  MSG_ACCEPTOR_ACCEPT = 1 # 追随者接受
  MSG_ACCEPTOR_REJECT = 2 # 追随者拒绝-网络不通
  MSG_ACCEPTOR_UNACCEPT = 3 # 追随者网络通-不同意
  MSG_ACCEPT = 4 # 接受
  MSG_PROPOSE = 5 # 提议
  MSG_EXT_PROPOSE = 6 # 额外提议
  MSG_HEARTBEAT = 7 # 心跳，每隔一段时间同步消息
  def __init__(self, command=None):
    self.command = command
  # 把收到的消息原原路返回，作为应答消息
  def copyAsReply(self, message):
    # 提议ID #当前的ID #发给谁 #谁发的
    self.proposalID, self.instanceID, self.to, self.source = message.proposalID, message.instanceID, message.source, message.to
    self.value = message.value # 发的信息

# 基于socket传递消息，封装网络传递消息
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: # 只要所有者线程没有结束,一直接受消息
        try:
          (bytes, addr) = self.owner.socket.recvfrom(2048) # 收取消息
          msg = pickle.loads(bytes) # 读取二进制数据转化为消息
          msg.source = addr[1]
          self.owner.queue.put(msg) # 队列存入消息
        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) # 通信参数
    self.socket.bind(("localhost", port)) # 通信地址，ip，端口
    self.socket.settimeout(timeout) # 超时设置
    self.queue = queue.Queue() # 队列
    self.helper = MessagePump.MPHelper(self) # 接收消息

  # 运行主线程
  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) # 抓取数据，最多等待3s
      return msg
    except:
      return None

  # 发送消息
  def sendMessage(self, message):
    bytes = pickle.dumps(message) # 转化为二进制
    address = ("localhost", message.to) # 地址ip,端口(ip,port)
    self.socket.sendto(bytes, address)
    return True
  #是否停止收取消息
  def doAbort(self):
    self.abort = True

from MessagePump import MessagePump
import random
class AdversarialMessagePump(MessagePump): # 类的继承
  # 对抗消息传输，延迟消息并任意顺序传递,模拟网络的延迟，消息传送并不是顺序
  def __init__(self, owner, port, timeout=2):
    MessagePump.__init__(self, owner, port, timeout) # 初始化父类
    self.messages = set() # 集合避免重复

  def waitForMessage(self):
    try:
      msg = self.queue.get(True, 0.1) # 从队列抓取数据
      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)) # 随机抓取消息发送
      self.messages.remove(msg) # 删除消息
    else:
      msg = None
    return msg