序
Lamport面包店算法是解决多个线程并发访问一个共享的单用户资源的互斥问题的算法。由莱斯利·兰波特发明。
算法类比
Lamport把这个并发控制算法非常直观地类比为顾客去面包店采购。
- 面包店一次只能接待一位顾客的采购。
- 已知有n位顾客要进入面包店采购，按照次序安排他们在前台登记一个签到号码。该签到号码逐次增加1。
- 顾客根据签到号码的由小到大的顺序依次入店购货。
- 完成购买的顾客在前台把其签到号码归0。如果完成购买的顾客要再次进店购买，就必须重新排队。
这个类比中的顾客就相当于线程，而入店购货就是进入临界区独占访问该共享资源。由于计算机实现的特点，存在两个线程获得相同的签到号码的情况，这是因为两个线程几乎同时申请排队的签到号码，读取已经发出去的签到号码情况，这两个线程读到的数据是完全一样的，然后各自在读到的数据上找到最大值，再加1作为自己的排队签到号码。
为此，该算法规定如果两个线程的排队签到号码相等，则线程id号较小的具有优先权。
原理
Lamport时间戳原理如下：
- 每个事件对应一个Lamport时间戳，初始值为0
- 如果事件在节点内发生，时间戳加1
- 如果事件属于发送事件，时间戳加1并在消息中带上该时间戳
- 如果事件属于接收事件，时间戳 = Max(本地时间戳，消息中的时间戳) + 1
5个原则
- 为了请求资源，进程A发送消息(Tm:A)给所有的其他进程，并且把这个消息放到进程队列中，Tm是消息的时间戳
- 当进程B接收到了进程A的(Tm:A)请求后，会把它放到自己的请求队列，然后发送一个带时间戳的确认消息给A
- 为了释放资源，进程A移除所有(Tm:A)的请求消息，然后发送带时间戳的A释放资源请求消息给其他所有的进程
- 当进程B接收到进程A释放资源的请求，它会移除队列中任意的(Tm:A)的资源请求
- 当满足以下两个条件时，进程A会被分配该资源：
- a)有一个(Tm:A)的请求，按照=>关系排在队列第一位；
- b)A接收到了一个时间戳大于Tm的来自所有其他进程的消息
代码示例
private void processRevcMsg(Message m) throws InterruptedException {
// 原理4 如果事件属于接收事件，时间戳 = Max(本地时间戳，消息中的时间戳) + 1
clock.update(m.getTimestamp());
lastSendMap.put(m.getFrom(), m);
switch (m.getMsgType()) {
case REQUEST_RES:
// rule 2 当进程B接收到了进程A的(Tm:A)请求后，会把它放到自己的请求队列，然后发送一个带时间戳的确认消息给A
addMessageToReqMap(m);
Message ackMsg = new Message(pid, m.getMsgId(), MessageType.REQUEST_ACK, clock.time());
// send ack to sender
sendToTargetProcess(ackMsg,m.getFrom());
break;
case REQUEST_ACK:
break;
case RELEASE_RES:
// rule 4 当进程B接收到进程A释放资源的请求，它会移除队列中任意的(Tm:A)的资源请求
dropMessageFromReqMap(m);
break;
default:
break;
}
tryToAcquireResource();
}
private void tryToAcquireResource() {
synchronized (reqMap) {
if(!reqMap.containsKey(pid) || reqMap.get(pid).isEmpty()){
return ;
}
Message myMessage = reqMap.get(pid).get(0);
int acceptCount = 1;
// rule 5 当满足以下两个条件时，进程A会被分配该资源：a)有一个(Tm:A)的请求，按照=>关系排在队列第一位；b)A接收到了一个时间戳大于Tm的来自所有其他进程的消息
// condition (ii) of rule 5
// A接收到了一个来自所有其他进程的消息,而且时间戳大于Tm
for (Map.Entry<Integer, Message> entry : lastSendMap.entrySet()) {
if (entry.getKey() == pid) {
continue;
}
if (isFirstEarlier(myMessage, entry.getValue())) {
acceptCount++;
}else{
return ;
}
}
if (!coordinator.hasAcceptedAll(acceptCount)){
return;
}
// condition (i) of rule 5
// 有一个Tm:A的请求，按照=>关系排在队列第一位
for (Map.Entry<Integer, List<Message>> entry : reqMap.entrySet()) {
if (entry.getKey() != pid && !entry.getValue().isEmpty()) {
if (!isFirstEarlier(myMessage, entry.getValue().get(0))) {
return;
}
}
}
// remove this request message
final Message firstMsg = reqMap.get(pid).remove(0);
workingPool.execute(new Runnable() {
public void run() {
coordinator.acquire(firstMsg.getMsgId(), pid, firstMsg.getTimestamp());
// emulate owning resources for a long time
try {
Thread.sleep(50L);
// rule 3 为了释放资源，进程A移除所有(Tm:A)的请求消息，然后发送带时间戳的A释放资源请求消息给其他所有的进程程
coordinator.release(firstMsg.getMsgId(), pid, firstMsg.getTimestamp());
Message releaseMsg = new Message(pid, firstMsg.getMsgId(),MessageType.RELEASE_RES, clock.time());
sendToOtherProcesses(releaseMsg);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
}
}
