摘要：序面包店算法是解決多個線程并發(fā)訪問一個共享的單用戶資源的互斥問題的算法。面包店一次只能接待一位顧客的采購。已知有位顧客要進入面包店采購，按照次序安排他們在前臺登記一個簽到號碼。顧客根據(jù)簽到號碼的由小到大的順序依次入店購貨。

Lamport面包店算法是解決多個線程并發(fā)訪問一個共享的單用戶資源的互斥問題的算法。由萊斯利·蘭波特發(fā)明。

Lamport把這個并發(fā)控制算法非常直觀地類比為顧客去面包店采購。

面包店一次只能接待一位顧客的采購。

已知有n位顧客要進入面包店采購，按照次序安排他們在前臺登記一個簽到號碼。該簽到號碼逐次增加1。

顧客根據(jù)簽到號碼的由小到大的順序依次入店購貨。

完成購買的顧客在前臺把其簽到號碼歸0。如果完成購買的顧客要再次進店購買，就必須重新排隊。

這個類比中的顧客就相當(dāng)于線程，而入店購貨就是進入臨界區(qū)獨占訪問該共享資源。由于計算機實現(xiàn)的特點，存在兩個線程獲得相同的簽到號碼的情況，這是因為兩個線程幾乎同時申請排隊的簽到號碼，讀取已經(jīng)發(fā)出去的簽到號碼情況，這兩個線程讀到的數(shù)據(jù)是完全一樣的，然后各自在讀到的數(shù)據(jù)上找到最大值，再加1作為自己的排隊簽到號碼。

為此，該算法規(guī)定如果兩個線程的排隊簽到號碼相等，則線程id號較小的具有優(yōu)先權(quán)。

Lamport時間戳原理如下：

每個事件對應(yīng)一個Lamport時間戳，初始值為0

如果事件在節(jié)點內(nèi)發(fā)生，時間戳加1

如果事件屬于發(fā)送事件，時間戳加1并在消息中帶上該時間戳

如果事件屬于接收事件，時間戳 = Max(本地時間戳，消息中的時間戳) + 1

為了請求資源，進程A發(fā)送消息(Tm:A)給所有的其他進程，并且把這個消息放到進程隊列中，Tm是消息的時間戳

當(dāng)進程B接收到了進程A的(Tm:A)請求后，會把它放到自己的請求隊列，然后發(fā)送一個帶時間戳的確認消息給A

為了釋放資源，進程A移除所有(Tm:A)的請求消息，然后發(fā)送帶時間戳的A釋放資源請求消息給其他所有的進程

當(dāng)進程B接收到進程A釋放資源的請求，它會移除隊列中任意的(Tm:A)的資源請求

當(dāng)滿足以下兩個條件時，進程A會被分配該資源：

a)有一個(Tm:A)的請求，按照=>關(guān)系排在隊列第一位；

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 當(dāng)進程B接收到了進程A的(Tm:A)請求后，會把它放到自己的請求隊列，然后發(fā)送一個帶時間戳的確認消息給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 當(dāng)進程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 當(dāng)滿足以下兩個條件時，進程A會被分配該資源：a)有一個(Tm:A)的請求，按照=>關(guān)系排在隊列第一位；b)A接收到了一個時間戳大于Tm的來自所有其他進程的消息

            // condition (ii) of rule 5
            // A接收到了一個來自所有其他進程的消息,而且時間戳大于Tm
            for (Map.Entry 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的請求，按照=>關(guān)系排在隊列第一位
            for (Map.Entry> 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)的請求消息，然后發(fā)送帶時間戳的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();
                    }

                }
            });
        }
    }

Lamport面包店算法

分布式系統(tǒng)理論基礎(chǔ) - 時間、時鐘和事件順序

分布式系統(tǒng)時序基礎(chǔ)

lamport