您现在的位置是:首页 > 正文

paxos算法java实现_基于python的Paxos算法实现

2024-01-30 22:24:41阅读 0

理解一个算法最快,最深刻的做法,我觉着可能是自己手动实现,虽然项目中不用自己实现,有已经封装好的算法库,供我们调用,我觉着还是有必要自己亲自实践一下。

这里首先说明一下,python这种动态语言,对不熟悉的人可能看着比较别扭,不像java那样参数类型是固定的,所以看着会有些蛋疼。这里环境用的是python2.7。

00276188264bb0ca27d74f17da9896b3.png

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上看这个逻辑,可以快速定位参数指向,如果有不对的地方欢迎指正

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持脚本之家。

网站文章

  • android studio 集成ffmpeg_android下vulkan与opengles纹理互通

    android studio 集成ffmpeg_android下vulkan与opengles纹理互通

    先放demo源码地址:https://github.com/xxxzhou/aoce 06_mediaplayer效果图:主要几个点:用ffmpeg打开rtmp流。使用vulkan Compute s...

    2024-01-30 22:24:11
  • Babel转码器入门

    Babel转码器入门

    Babel转码

    2024-01-30 22:24:05
  • 谈谈代码中的this

      js中我们常常会遇到this,this的具体指向问题对于很多同学来说是很懵懂的;就想lz刚开始接触时候就是一脸的懵逼,经常被一些题目转的眼花缭乱。那么今天lz就跟大家一起交流一下这个this的指向问题!  背景:不久前有个同事给我们发了一道有关this的题目,于是便有了今天的故事。如下题:class D0XX { constructor () { this...

    2024-01-30 22:23:58
  • vue3 开启 https

    vue3 开启 https

    会在当前目录生成,cert.crt 和 cert.key 两个文件。会在当前目录生成,ca.crt 和 ca.key 两个文件。

    2024-01-30 22:23:29
  • 软件工程≠计算机科学 之 思考

    软件工程≠计算机科学首先,这篇文章的英文比上次阅读作业的英文好懂多了可能是因为上次的多是讲技术的,可以归类为科技英语吧,好多的词都不认识这次更偏向于议论文吧We need software that can be put together in a reasonable amount of time, for a reasonable cost, that works som...

    2024-01-30 22:23:21
  • 用电脑开热点启用电脑无线配置

    用电脑开热点启用电脑无线配置

    电脑系统打开热点,不用安装360也能用电脑给手机开热点,给多电脑开热点无线。

    2024-01-30 22:23:16
  • 【算法竞赛学习】数字中国创新大赛智慧海洋建设-Task3特征工程

    智慧海洋建设-Task3 特征工程此部分为智慧海洋建设竞赛的特征工程模块,通过特征工程,可以最大限度地从原始数据中提取特征以供算法和模型使用。通俗而言,就是通过X,创造新的X’以获得更好的训练、预测效...

    2024-01-30 22:22:46
  • 进阶面向对象

    进阶面向对象

    对象是具体的事物,拥有所属类中描述的一切属性和行为。类是一种模型,这种模型可以创建出不同的对象实体。一个类可以有很多对象,而一个对象必然属于某个类。类用于抽象的描述一类事物所特有的属性和行为。将日常生...

    2024-01-30 22:22:39
  • 沈阳师范大学大一上册C语言PTA题目集以及答案(第六章 指针 编程题篇)

    沈阳师范大学大一上册C语言PTA题目集以及答案(第六章 指针 编程题篇)7-1 输出学生成绩 (20分)本题要求编写程序,根据输入学生的成绩,统计并输出学生的平均成绩、最高成绩和最低成绩。建议使用动态...

    2024-01-30 22:22:32
  • 常用的CMD命令+Python

    cd .. 返回上一级目录 cd ../.. 返回上两级目录 cd或cd ~ 返回home目录 cd - 目录名 返回指定目录 输入“cd\”敲回车会直接退回到当前盘根目录 在CMD程序里输入“d:”...

    2024-01-30 22:22:23