基于python的Paxos算法实现
理解一个算法最快,最深刻的做法,我觉着可能是自己手动实现,虽然项目中不用自己实现,有已经封装好的算法库,供我们调用,我觉着还是有必要自己亲自实践一下。
这里首先说明一下,python这种动态语言,对不熟悉的人可能看着比较别扭,不像java那样参数类型是固定的,所以看着会有些蛋疼。这里环境用的是python2.7。
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,线程和队列实现的消息处理器:
# 基于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
再来一个是记录类
# InstanceRecord本地记录类,主要记录追随者、领导者最高编号的协议 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 # 取得编号最大的协议 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("删除协议") del self.protocols[k]
下面就是Acceptor的实现:
# 追随者 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 # 发送消息 def sendMessage(self, message): self.msgPump.sendMessage(message) # 收消息,只收取为提议的消息 def recvMessage(self, message): if message == None: return if self.failed: # 失败状态不收取消息 return if message.command == Message.MSG_PROPOSE: # 判断消息是否为提议 if message.instanceID not in self.instances: record = InstanceRecord() # 记录器 self.instances[message.instanceID] = record protocol = PaxosAcceptorProtocol(self) # 创建协议 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 # 储存信息 print(u"协议被客户端接受 %s" % message.value) # 获取最高同意的建议 def getHighestAgreedProposal(self, instance): return self.instances[instance].highestID # (port,count) # 获取接口数据 def getInstanceValue(self, instance): return self.instances[instance].value
那再看下AcceptorProtocol的实现:
from Message import Message class PaxosAcceptorProtocol(object): # State variables STATE_UNDEFINED = -1 # 协议没有定义的情况0 STATE_PROPOSAL_RECEIVED = 0 # 收到消息 STATE_PROPOSAL_REJECTED = 1 # 拒绝链接 STATE_PROPOSAL_AGREED = 2 # 同意链接 STATE_PROPOSAL_ACCEPTED = 3 # 同意请求 STATE_PROPOSAL_UNACCEPTED = 4 # 拒绝请求 def __init__(self, client): self.client = client self.state = PaxosAcceptorProtocol.STATE_UNDEFINED # 收取,只处理协议类型的消息 def recvProposal(self, message): if message.command == Message.MSG_PROPOSE: # 协议 self.proposalID = message.proposalID self.instanceID = message.instanceID (port, count) = self.client.getHighestAgreedProposal(message.instanceID) # 端口,协议内容的最高编号 # 检测编号处理消息协议 # 判断协议是否最高 if count < self.proposalID[1] or (count == self.proposalID[1] and port < self.proposalID[0]): self.state = PaxosAcceptorProtocol.STATE_PROPOSAL_AGREED # 协议同意 print("同意协议:%s, %s " % (message.instanceID, message.value)) value = self.client.getInstanceValue(message.instanceID) msg = Message(Message.MSG_ACCEPTOR_AGREE) # 同意协议 msg.copyAsReply(message) msg.value = value msg.sequence = (port, count) self.client.sendMessage(msg) # 发送消息 else: # 不再接受比最高协议小的提议 self.state = PaxosAcceptorProtocol.STATE_PROPOSAL_REJECTED return self.proposalID else: # 错误重试 pass # 过度 def doTransition(self, message): # 如果当前协议状态是接受连接,消息类型是接受 if self.state == PaxosAcceptorProtocol.STATE_PROPOSAL_AGREED and message.command == Message.MSG_ACCEPT: self.state = PaxosAcceptorProtocol.STATE_PROPOSAL_ACCEPTED # 接收协议 msg = Message(Message.MSG_ACCEPTOR_ACCEPT) # 创造消息 msg.copyAsReply(message) # 拷贝并回复 for l in self.client.leaders: msg.to = l self.client.sendMessage(msg) # 给领导发送消息 self.notifyClient(message) # 通知自己 return True raise Exception("并非预期的状态和命令") # 通知 自己客户端 def notifyClient(self, message): self.client.notifyClient(self, message)
接着看下Leader和LeaderProtocol实现:
# 领导者 import threading import Queue import time from Message import Message from MessagePump import MessagePump from InstanceRecord import InstanceRecord from PaxosLeaderProtocol import PaxosLeaderProtocol class PaxosLeader: # 定时监听 class HeartbeatListener(threading.Thread): def __init__(self, leader): self.leader = leader self.queue = Queue.Queue() # 消息队列 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) # 设定规则,谁的端口号比较高,谁就是领导 if hb.source > self.leader.port: self.leader.setPrimary(False) except: self.leader.setPrimary(True) # 定时发送 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 # 自身是不是领导 self.proposalCount = 0 self.msgPump = MessagePump(self, port) # 消息传送器 self.instances = {} self.hbListener = PaxosLeader.HeartbeatListener(self) # 监听 self.hbSender = PaxosLeader.HeartbeatSender(self) # 发送心跳 self.highestInstance = -1 # 协议状态 self.stoped = True # 是否正在运行 self.lasttime = time.time() # 最后一次时间 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): # 设置领导者 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 # 获取所有的领导下面的追随者 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)] # 抓取同意的数量 def getNumAccpted(self): return len([v for v in self.getHistory() if v != None]) # 抓取空白时间处理下事务 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() # 通知领导 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("额外的协议", 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:#创建协议 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: # 创建协议标号 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("协议接口%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实现:
from Message import Message class PaxosLeaderProtocol(object): STATE_UNDEFINED = -1 # 协议没有定义的情况0 STATE_PROPOSED = 0 # 协议消息 STATE_REJECTED = 1 # 拒绝链接 STATE_AGREED = 2 # 同意链接 STATE_ACCEPTED = 3 # 同意请求 STATE_UNACCEPTED = 4 # 拒绝请求 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): # 根據狀態運行協議 if self.state == PaxosLeaderProtocol.STATE_PROPOSED: if message.command == Message.MSG_ACCEPTOR_AGREE: self.agreecount += 1 if self.agreecount >= self.leader.getQuorumSize(): # 选举 print(u"达成协议的法定人数,最后的价值回答是:%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 # 同意更新 # 发送同意消息 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: # 同意协议 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)
测试模块:
import socket, pickle, time from Message import Message from PaxosAcceptor import PaxosAcceptor from PaxosLeader import PaxosLeader if __name__ == "__main__": # 设定5个客户端 numclients = 5 clients = [PaxosAcceptor(port, [54321, 54322]) for port in range(64320, 64320 + numclients)] # 两个领导者 leader1 = PaxosLeader(54321, [54322], [c.port for c in clients]) leader2 = PaxosLeader(54322, [54321], [c.port for c in clients]) # 开启领导者与追随者 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协议 start = time.time() for i in range(1000): m = Message(Message.MSG_EXT_PROPOSE) # 消息 m.value = 0 + i # 消息参数 m.to = 54322 # 设置传递的端口 bytes = pickle.dumps(m) # 提取的二进制数据 s.sendto(bytes, ("localhost", m.to)) # 发送消息 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历史纪录") print(leader1.getHistory()) print(u"leader2历史纪录") print(leader2.getHistory()) end = time.time() print(u"一共用了%f秒" % (end - start))
代码确实比较长,看起来有些困难,最好还是在pycharm上看这个逻辑,可以快速定位参数指向,如果有不对的地方欢迎指正
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持我们。
赞 (0)