摘要：文章結(jié)構(gòu)來自七周七并發(fā)模型互斥和內(nèi)存模型創(chuàng)建線程這段代碼創(chuàng)建并啟動了一個實例，首先從開始，函數(shù)的余下部分一起并發(fā)執(zhí)行。在鎖定狀態(tài)下，某些線程擁有鎖在非鎖定狀態(tài)下，沒有線程擁有它。

并發(fā)程序含有多個邏輯上的獨立執(zhí)行塊，他們可以獨立的并行執(zhí)行，也可以串行執(zhí)行。
并行程序解決問題的速度比串行程序快的多，因為其可以同時執(zhí)行整個任務(wù)的多個部分。并行程序可能有多個獨立執(zhí)行塊，也可能只有一個。

引用Rob Pike的經(jīng)典描述就是：
并發(fā)是同一時間應(yīng)對多件事情的能力；
并行是同一時間動手做多件事情的能力。

常見的并發(fā)模型有：

線程與鎖

函數(shù)式編程

actor模型和通信順序是進行（Communicating Sequential Processes, CSP）

數(shù)據(jù)級并行

lambda 架構(gòu)

分離標(biāo)識與狀態(tài)模型

這篇主要介紹線程與鎖模型

線程與鎖模型是對底層硬件運行過程的形式化，非常簡單直接，幾乎所有的編程語言都對其提供了支持，且不對其使用方法加以限制（易出錯）。

這篇文章主要使用python語言來演示線程與鎖模型。文章結(jié)構(gòu)來自《七周七并發(fā)模型》

from threading import Thread

def hello_world():
    print("Hello from new thread")

def main():
    my_thread = Thread(target=hello_world)
    my_thread.start()
    print("Hello from main thread")
    my_thread.join()

main()

這段代碼創(chuàng)建并啟動了一個Thread實例，首先從start() 開始，my_thread.start() main()函數(shù)的余下部分一起并發(fā)執(zhí)行。最后調(diào)用join() 來等待my_thread線程結(jié)束。

運行這段代碼輸出結(jié)果有幾種：

Hello from new thread
Hello from main thread

或者

Hello from main thread
Hello from new thread

或者

Hello from new threadHello from main thread

究竟哪個結(jié)果取決于哪個線程先執(zhí)行print()。多線程編程很難的原因之一就是運行結(jié)果可能依賴于時序，多次運行結(jié)果并不穩(wěn)定。

from threading import Thread, Lock


class Counter(object):
    def __init__(self, count=0):
        self.count = count

    def increment(self):
        self.count += 1

    def get_count(self):
        print("Count: %s" % self.count)
        return self.count


def test_count():
    counter = Counter()

    class CounterThread(Thread):
        def run(self):
            for i in range(10000):
                counter.increment()

    t1 = CounterThread()
    t2 = CounterThread()
    t1.start()
    t2.start()
    t1.join()
    t2.join()

    counter.get_count()

test_count()

這段代碼創(chuàng)建一個簡單的類Counter 和兩個線程，每個線程都調(diào)用counter.increment() 10000次。

多次運行這段代碼會得到不同的值，原因是兩個線程在使用 counter.count 時發(fā)生了競態(tài)條件（代碼行為取決于各操作的時序）。

一個可能的操作是：

線程t1 獲取count的值時，線程t2也同時獲取到了count 的值（假設(shè)是100），

這時t1 count + 1， 此時count 為101，回寫count 值，然后t2 執(zhí)行了相同的操作 count+1，因為t2 取到的值也是100 此時 count 仍是101，回寫后count 依然是101，但是 +1 操作執(zhí)行了兩次。

競態(tài)條件的解決方法是對 count 進行同步（synchronize）訪問。一種操作是使用 內(nèi)置鎖(也稱互斥鎖（mutex）、管程（monitor）或臨界區(qū)（critical section）)來同步對increment() 的調(diào)用。

線程同步能夠保證多個線程安全訪問競爭資源，最簡單的同步機制是引入互斥鎖。互斥鎖為資源引入一個狀態(tài)：鎖定/非鎖定。某個線程要更改共享數(shù)據(jù)時，先將其鎖定，此時資源的狀態(tài)為“鎖定”，其他線程不能更改；
直到該線程釋放資源，將資源的狀態(tài)變成“非鎖定”，其他的線程才能再次鎖定該資源。互斥鎖保證了每次只有一個線程進行寫入操作，從而保證了多線程情況下數(shù)據(jù)的正確性。
當(dāng)一個線程調(diào)用鎖的acquire()方法獲得鎖時，鎖就進入“l(fā)ocked”狀態(tài)。每次只有一個線程可以獲得鎖。如果此時另一個線程試圖獲得這個鎖，該線程就會變?yōu)椤癰locked”狀態(tài)，稱為“同步阻塞”。
直到擁有鎖的線程調(diào)用鎖的release()方法釋放鎖之后，鎖進入“unlocked”狀態(tài)。線程調(diào)度程序從處于同步阻塞狀態(tài)的線程中選擇一個來獲得鎖，并使得該線程進入運行（running）狀態(tài)。

python 鎖的使用流程如下：

#創(chuàng)建鎖
mutex = threading.Lock()
#鎖定
mutex.acquire([timeout])
#釋放
mutex.release()

推薦使用上下文管理器來操作鎖，

with lock:
    do someting

# 相當(dāng)于

lock.acquire()

try:
    # do something...
finally:
    lock.release()

acquire(blocking=True, timeout=-1)
可以阻塞或非阻塞地獲得鎖。
當(dāng)調(diào)用時參數(shù) blocking 設(shè)置為 True （缺省值），阻塞直到鎖被釋放，然后將鎖鎖定并返回 True 。
在參數(shù) blocking 被設(shè)置為 False 的情況下調(diào)用，將不會發(fā)生阻塞。如果調(diào)用時 blocking 設(shè)為 True 會阻塞，并立即返回 False ；否則，將鎖鎖定并返回 True。
當(dāng)浮點型 timeout 參數(shù)被設(shè)置為正值調(diào)用時，只要無法獲得鎖，將最多阻塞 timeout 設(shè)定的秒數(shù)。timeout 參數(shù)被設(shè)置為 -1 時將無限等待。當(dāng) blocking 為 false 時，timeout 指定的值將被忽略。
如果成功獲得鎖，則返回 True，否則返回 False (例如發(fā)生 超時 的時候)。
timeout 參數(shù)需要 python3.2+

from threading import Thread, Lock
mutex = Lock()

class SynchronizeCounter(object):
    def __init__(self, count=0):
        self.count = count

    def increment(self):
        # if mutex.acquire(1): # 獲取鎖
        #     self.count += 1
        #     mutex.release()  # 釋放鎖
        # 等同于上述代碼
        with mutex:
            self.count += 1

    def get_count(self):
        print("Count: %s" % self.count)
        return self.count

def test_synchronize_count():
    counter = SynchronizeCounter()

    class CounterThread(Thread):
        def run(self):
            for i in range(100000):
                counter.increment()

    t1 = CounterThread()
    t2 = CounterThread()
    t1.start()
    t2.start()
    t1.join()
    t2.join()

    counter.get_count()


if __name__ == "__main__":
    for i in range(100):
        test_synchronize_count()

這段代碼還有一個隱藏的bug，那就是 get_count()，這里get_count() 是在join()之后調(diào)用的，因此是線程安全的，但是如果在其它地方調(diào)用了 get_count() 函數(shù)。
由于在 get_count() 中沒有進行線程同步，調(diào)用時可能會獲取到一個失效的值。

對于JAVA等競態(tài)編譯語言，

編譯器的靜態(tài)優(yōu)化可能會打亂代碼的執(zhí)行順序

JVM 的動態(tài)優(yōu)化也會打亂代碼的執(zhí)行順序

硬件可以通過亂序執(zhí)行來優(yōu)化性能

更糟糕的是，有時一個線程產(chǎn)生的修改可能會對另一個線程不可見。

從直覺上來說，編譯器、JVM、硬件都不應(yīng)插手修改原本的代碼邏輯。但是近幾年的運行效率提升，尤其是共享內(nèi)存交媾的運行效率提升，都仰仗于此類代碼優(yōu)化。
具體的副作用，Java 內(nèi)存模型有明確說明。
Java 內(nèi)存模型定義了何時一個線程對內(nèi)存的修改對另一個線程可見。基本原則是：如果讀線程和寫線程不進行同步，就不能保證可見性。

一個重點： 兩個線程都需要進行同步。只在其中一個線程進行同步是不夠的。

可如果所有的方法都同步，大多數(shù)線程可能都會被阻塞，失去了并發(fā)的意義，并且可能會出現(xiàn)死鎖。

哲學(xué)家進餐問題

哲學(xué)家就餐問題：假設(shè)有五位哲學(xué)家圍坐在一張圓形餐桌旁，做以下兩件事情之一：吃飯，或者思考。吃東西的時候，他們就停止思考，思考的時候也停止吃東西。餐桌中間有一大碗意大利面，每兩個哲學(xué)家之間有一只餐叉。因為用一只餐叉很難吃到意大利面，所以假設(shè)哲學(xué)家必須用兩只餐叉吃東西。他們只能使用自己左右手邊的那兩只餐叉。 
哲學(xué)家從來不交談，這就很危險，可能產(chǎn)生死鎖，每個哲學(xué)家都拿著左手的餐叉，永遠都在等右邊的餐叉（或者相反）。
即使沒有死鎖，也有可能發(fā)生資源耗盡。例如，假設(shè)規(guī)定當(dāng)哲學(xué)家等待另一只餐叉超過五分鐘后就放下自己手里的那一只餐叉，并且再等五分鐘后進行下一次嘗試。這個策略消除了死鎖（系統(tǒng)總會進入到下一個狀態(tài)），但仍然有可能發(fā)生“活鎖”。如果五位哲學(xué)家在完全相同的時刻進入餐廳，并同時拿起左邊的餐叉，那么這些哲學(xué)家就會等待五分鐘，同時放下手中的餐叉，再等五分鐘，又同時拿起這些餐叉。

下面是哲學(xué)家進餐問題的一個實現(xiàn)：

import threading
import random
import time


class Philosopher(threading.Thread):

    running = True

    def __init__(self, xname, forkOnLeft, forkOnRight):
        threading.Thread.__init__(self)
        self.name = xname
        self.forkOnLeft = forkOnLeft
        self.forkOnRight = forkOnRight

    def run(self):
        while self.running:
            #  Philosopher is thinking (but really is sleeping).
            time.sleep(random.uniform(1, 3))
            print("%s is hungry." % self.name)
            self.dine()

    def dine(self):
        fork1, fork2 = self.forkOnLeft, self.forkOnRight

        while self.running:
            fork1.acquire(True)  # 阻塞式獲取left 鎖
            # locked = fork2.acquire(True) # 阻塞式 獲取right 鎖  容易產(chǎn)生死鎖
            locked = fork2.acquire(False)  # 非阻塞式 獲取right 鎖
            if locked:
                break  # 如果被鎖定，釋放 left 退出等待
            fork1.release()
            print("%s swaps forks" % self.name)
            fork1, fork2 = fork2, fork1
        else:
            return

        self.dining()
        fork2.release()
        fork1.release()

    def dining(self):
        print("%s starts eating " % self.name)
        time.sleep(random.uniform(1, 5))
        print("%s finishes eating and leaves to think." % self.name)


def DiningPhilosophers():
    forks = [threading.Lock() for n in range(5)]
    philosopherNames = ("Aristotle", "Kant", "Buddha", "Marx", "Russel")

    philosophers = [
        Philosopher(philosopherNames[i], forks[i % 5], forks[(i + 1) % 5])
        for i in range(5)
    ]

    Philosopher.running = True
    for p in philosophers:
        p.start()
    for p in philosophers:
        p.join()
    time.sleep(100)
    Philosopher.running = False
    print("Now we"re finishing.")


DiningPhilosophers()

規(guī)模較大的程序常用監(jiān)聽器模式來解耦模塊，這里我們構(gòu)造一個類從一個URL進行下載，Listeners 監(jiān)聽下載進度。

import requests
import threading


class Listeners(object):
    def __init__(self, count=0):
        self.count = count
        self.done_count = 0.0
        self.listeners = []

    def append(self, listener):
        self.listeners.append(listener)

    def remove(self, listener):
        self.listeners.remove(listener)

    def on_progress(self, n):
        # 一些我們不知道的實現(xiàn)
        # do someting
        # self.done_count += 1
        # print("Process: %f" % (self.done_count / self.count))
        pass


listeners = Listeners(5)


class Downloader(threading.Thread):
    def __init__(
        self, group=None, target=None, name=None, args=(), kwargs=None, daemon=None
    ):
        threading.Thread.__init__(
            self, group=group, target=target, name=name, daemon=daemon
        )
        self.url = kwargs.get("url")

    def download(self):
        resp = requests.get(self.url)

    def add_listener(self, listener):
        listeners.append(listener)

    def remove_listener(self, listener):
        listeners.delete(listener)

    def update_progress(self, n):
        for listener in listeners:
            listner.on_progress(n)

    def run(self):
        self.download()
        print(self.url)
        listeners.on_progress(1)


def test():
    urls = [
        "https://www.baidu.com",
        "https://www.google.com",
        "https://www.bing.com",
        "https://www.zaih.com",
        "https://www.github.com",
    ]
    ts = [Downloader(kwargs=dict(url=url)) for url in urls]
    print(ts)
    [t.start() for t in ts]
    [t.join() for t in ts]


if __name__ == "__main__":
    test()

這段代碼中，add_listener， remove_listener 和 update_progress 都是同步方法，但 update_progress 調(diào)用了一個我們不知道如何實現(xiàn)的方法。如果這個方法中，獲取了一把鎖，程序在執(zhí)行的過程中就可能發(fā)生死鎖。所以，我們要盡量避免使用這種方法。還有一種方法是在遍歷之前對 listeners 進行保護性復(fù)制，再針對這份副本進行遍歷。（現(xiàn)在調(diào)用外星方法不再需要加鎖）

Lock() 雖然方便，但限制很多：

一個線程因為等待內(nèi)置鎖而進入阻塞之后，就無法中斷該線程

Lock() 不知道當(dāng)前擁有鎖的線程是否是當(dāng)前線程，如果當(dāng)前線程獲取了鎖，再次獲取也會阻塞。

重入鎖是(threading.RLock)一個可以被同一個線程多次獲取的同步基元組件。在內(nèi)部，它在基元鎖的鎖定/非鎖定狀態(tài)上附加了 "所屬線程" 和 "遞歸等級" 的概念。在鎖定狀態(tài)下，某些線程擁有鎖 ； 在非鎖定狀態(tài)下， 沒有線程擁有它。

若要鎖定鎖，線程調(diào)用其 acquire() 方法；一旦線程擁有了鎖，方法將返回。若要解鎖，線程調(diào)用 release() 方法。 acquire()/release() 對可以嵌套；只有最終 release() (最外面一對的 release() ) 將鎖解開，才能讓其他線程繼續(xù)處理 acquire() 阻塞。

threading.RLock 提供了顯式的 acquire() 和 release() 方法
一個好的實踐是：

lock = threading.RLock()

Lock 和 RLock 的使用區(qū)別如下：

#rlock_tut.py
import threading
num = 0
lock = Threading.Lock()
lock.acquire()
num += 1
lock.acquire() # 這里會被阻塞
num += 2
lock.release()

# With RLock, that problem doesn’t happen.
lock = Threading.RLock()
lock.acquire()
num += 3
lock.acquire() # 不會被阻塞.
num += 4
lock.release()
lock.release() # 兩個鎖都需要調(diào)用 release() 來釋放.

使用內(nèi)置鎖時，阻塞的線程無法被中斷，程序不能從死鎖恢復(fù)，可以給鎖設(shè)置超時時間來解決這個問題。

timeout 參數(shù)需要 python3.2+

import time
from threading import Thread, Lock

lock1 = RLock()
lock2 = RLock()

# 這個程序會一直死鎖下去，如果想突破這個限制，可以在獲取鎖的時候加上超時時間
# > python threading 沒有實現(xiàn) 銷毀(destroy)，停止(stop)，暫停(suspend)，繼續(xù)（resume）,中斷（interrupt）等


class T1(Thread):
    def run(self):
        print("start run T1")
        lock1.acquire()
        # lock1.acquire(timeout=2) # 設(shè)置超時時間可避免死鎖
        time.sleep(1)
        lock2.acquire()
        # lock2.acquire(timeout=2) # 設(shè)置超時時間可避免死鎖
        lock1.release()
        lock2.release()


class T2(Thread):
    def run(self):
        print("start run T2")
        lock2.acquire()
        # lock2.acquire(timeout=2) # 設(shè)置超時時間可避免死鎖
        time.sleep(1)
        lock1.acquire()
        # lock1.acquire(timeout=2) # 設(shè)置超時時間可避免死鎖
        lock2.release()
        lock1.release()


def test():
    t1, t2 = T1(), T2()
    t1.start()
    t2.start()

    t1.join()
    t2.join()


if __name__ == "__main__":
    test()

如果我們要在鏈表中插入一個節(jié)點。一種做法是用鎖保護整個鏈表，但鏈表加鎖時其它使用者無法訪問。交替鎖可以只所追殺鏈表的一部分，允許不涉及被鎖部分的其它線程自由訪問。

from random import randint
from threading import Thread, Lock

class Node(object):

    def __init__(self, value, prev=None, next=None):
        self.value = value
        self.prev = prev
        self.next = next
        self.lock = Lock()


class SortedList(Thread):

    def __init__(self, head):
        Thread.__init__(self)
        self.head = head

    def insert(self, value):
        head = self.head
        node = Node(value)
        print("insert: %d" % value)
        while True:
            if head.value <= value:
                if head.next != None:
                    head = head.next
                else:
                    head.lock.acquire()
                    head.next = node
                    node.prev = head
                    head.lock.release()
                    break
            else:
                prev = head.prev
                prev.lock.acquire()
                head.lock.acquire()
                if prev != None:
                    prev.next = node
                else:
                    self.head = node
                node.prev = prev
                prev.lock.release()
                node.next = head
                head.prev = node
                head.lock.release()
                break

    def run(self):
        for i in range(5):
            self.insert(randint(10, 20))


def test():
    head = Node(10)
    t1 = SortedList(head)
    t2 = SortedList(head)
    t1.start()
    t2.start()
    t1.join()
    t2.join()
    while head:
        print(head.value)
        head = head.next


if __name__ == "__main__":
    test()

這種方案不僅可以讓多個線程并發(fā)的進行鏈表插入操作，還能讓其他的鏈表操作安全的并發(fā)。

并發(fā)編程經(jīng)常需要等待某個事件發(fā)生。比如從隊列刪除元素前需要等待隊列非空、向緩存添加數(shù)據(jù)前需要等待緩存有足夠的空間。條件變量就是為這種情況設(shè)計的。

條件變量總是與某種類型的鎖對象相關(guān)聯(lián)，鎖對象可以通過傳入獲得，或者在缺省的情況下自動創(chuàng)建。當(dāng)多個條件變量需要共享同一個鎖時，傳入一個鎖很有用。鎖是條件對象的一部分，不必多帶帶地跟蹤它。

條件變量服從上下文管理協(xié)議：使用 with 語句會在它包圍的代碼塊內(nèi)獲取關(guān)聯(lián)的鎖。 acquire() 和 release() 方法也能調(diào)用關(guān)聯(lián)鎖的相關(guān)方法。

其它方法必須在持有關(guān)聯(lián)的鎖的情況下調(diào)用。 wait() 方法釋放鎖，然后阻塞直到其它線程調(diào)用 notify() 方法或 notify_all() 方法喚醒它。一旦被喚醒， wait() 方法重新獲取鎖并返回。它也可以指定超時時間。

#condition_tut.py
import random, time
from threading import Condition, Thread
"""
"condition" variable will be used to represent the availability of a produced
item.
"""
condition = Condition()
box = []
def producer(box, nitems):
    for i in range(nitems):
        time.sleep(random.randrange(2, 5))  # Sleeps for some time.
        condition.acquire()
        num = random.randint(1, 10)
        box.append(num)  # Puts an item into box for consumption.
        condition.notify()  # Notifies the consumer about the availability.
        print("Produced:", num)
        condition.release()
def consumer(box, nitems):
    for i in range(nitems):
        condition.acquire()
        condition.wait()  # Blocks until an item is available for consumption.
        print("%s: Acquired: %s" % (time.ctime(), box.pop()))
        condition.release()
threads = []
"""
"nloops" is the number of times an item will be produced and
consumed.
"""
nloops = random.randrange(3, 6)
for func in [producer, consumer]:
    threads.append(Thread(target=func, args=(box, nloops)))
    threads[-1].start()  # Starts the thread.
for thread in threads:
    """Waits for the threads to complete before moving on
       with the main script.
    """
    thread.join()
print("All done.")

與鎖相比使用原子變量的優(yōu)點：

不會忘記在正確的時候獲取鎖
由于沒有鎖的參與，對原子變量的操作不會引發(fā)死鎖。
原子變量時無鎖（lock-free）非阻塞（non-blocking）算法的基礎(chǔ)，這種算法可以不用鎖和阻塞來達到同步的目的。

python 不支持原子變量

線程與鎖模型最大的優(yōu)點是適用面廣，更接近于“本質(zhì)”--近似于對硬件工作方式的形式化--正確使用時效率高。
此外，線程與鎖模型也可輕松的集成到大多數(shù)編程語言。

線程與鎖模型沒有為并行提供直接的支持

線程與鎖模型只支持共享內(nèi)存模型，如果要支持分布式內(nèi)存模型，就需要尋求其他技術(shù)的幫助。

用線程與鎖模型編寫的代碼難以測試（比如死鎖問題可能很久才會出現(xiàn)），出了問題后很難找到問題在哪，并且bug難以復(fù)現(xiàn)

代碼難以維護（要保證所有對象的同步都是正確的、必須按 順序來獲取多把鎖、持有鎖時不調(diào)用外星方法。還要保證維護代碼的開發(fā)者都遵守這個規(guī)則

Let’s Synchronize Threads in Python

哲學(xué)家進餐問題

[1] 哲學(xué)家進餐問題: https://zh.wikipedia.org/wiki...
[2] Let’s Synchronize Threads in Python: https://hackernoon.com/synchr...