理解 Python 中的多線程

時間 2019-11-15

原文原文鏈接

咱們將會看到一些在 Python 中使用線程的實例和如何避免線程之間的競爭。
你應當將下邊的例子運行屢次，以即可以注意到線程是不可預測的和線程每次運行出的不一樣結果。聲明：從這裏開始忘掉你聽到過的關於 GIL 的東西，由於 GIL 不會影響到我想要展現的東西。
html

示例1，咱們將要請求五個不一樣的url：

一、單線程

import time
import urllib2
 
def get_responses():
    urls = [
        'http://www.google.com',
        'http://www.amazon.com',
        'http://www.ebay.com',
        'http://www.alibaba.com',
        'http://www.reddit.com'
    ]
    start = time.time()
    for url in urls:
        print url
        resp = urllib2.urlopen(url)
        print resp.getcode()
    print "Elapsed time: %s" % (time.time()-start)
 
get_responses()

輸出是：

http://www.google.com 200
http://www.amazon.com 200
http://www.ebay.com 200
http://www.alibaba.com 200
http://www.reddit.com 200
python

Elapsed time: 3.0814409256 git

解釋：程序員

url順序的被請求
除非cpu從一個url得到了迴應，不然不會去請求下一個url
網絡請求會花費較長的時間，因此cpu在等待網絡請求的返回時間內一直處於閒置狀態。
github

二、多線程

import urllib2
import time
from threading import Thread
 
class GetUrlThread(Thread):
    def __init__(self, url):
        self.url = url 
        super(GetUrlThread, self).__init__()
 
    def run(self):
        resp = urllib2.urlopen(self.url)
        print self.url, resp.getcode()
 
def get_responses():
    urls = [
        'http://www.google.com', 
        'http://www.amazon.com', 
        'http://www.ebay.com', 
        'http://www.alibaba.com', 
        'http://www.reddit.com'
    ]
    start = time.time()
    threads = []
    for url in urls:
        t = GetUrlThread(url)
        threads.append(t)
        t.start()
    for t in threads:
        t.join()
    print "Elapsed time: %s" % (time.time()-start)
 
get_responses()

輸出：
http://www.reddit.com 200
http://www.google.com 200
http://www.amazon.com 200
http://www.alibaba.com 200
http://www.ebay.com 200
編程

Elapsed time: 0.689890861511 安全

解釋：
意識到了程序在執行時間上的提高
咱們寫了一個多線程程序來減小cpu的等待時間，當咱們在等待一個線程內的網絡請求返回時，這時cpu能夠切換到其餘線程去進行其餘線程內的網絡請求。
咱們指望一個線程處理一個url，因此實例化線程類的時候咱們傳了一個url。
線程運行意味着執行類裏的run()方法。
不管如何咱們想每一個線程必須執行run()。
爲每一個url建立一個線程而且調用start()方法，這告訴了cpu能夠執行線程中的run()方法了。
咱們但願全部的線程執行完畢的時候再計算花費的時間，因此調用了join()方法。
join()能夠通知主線程等待這個線程結束後，才能夠執行下一條指令。
每一個線程咱們都調用了join()方法，因此咱們是在全部線程執行完畢後計算的運行時間。
關於線程：
cpu可能不會在調用start()後立刻執行run()方法。
你不能肯定run()在不一樣線程建間的執行順序。
對於單獨的一個線程，能夠保證run()方法裏的語句是按照順序執行的。
這就是由於線程內的url會首先被請求，而後打印出返回的結果。
網絡

示例2，全局變量的線程安全問題（race condition）

一、BUG 版

咱們將會用一個程序演示一下多線程間的資源競爭，並修復這個問題。多線程

from threading import Thread
 
#define a global variable
some_var = 0
 
class IncrementThread(Thread):
    def run(self):
        #we want to read a global variable
        #and then increment it
        global some_var
        read_value = some_var
        print "some_var in %s is %d" % (self.name, read_value)
        some_var = read_value + 1
        print "some_var in %s after increment is %d" % (self.name, some_var)
 
def use_increment_thread():
    threads = []
    for i in range(50):
        t = IncrementThread()
        threads.append(t)
        t.start()
    for t in threads:
        t.join()
    print "After 50 modifications, some_var should have become 50"
    print "After 50 modifications, some_var is %d" % (some_var,)
 
use_increment_thread()

屢次運行這個程序，你會看到多種不一樣的結果。
解釋：
有一個全局變量，全部的線程都想修改它。
全部的線程應該在這個全局變量上加 1 。
有50個線程，最後這個數值應該變成50，可是它卻沒有。
爲何沒有達到50？
在some_var是15的時候，線程t1讀取了some_var，這個時刻cpu將控制權給了另外一個線程t2。
t2線程讀到的some_var也是15
t1和t2都把some_var加到16
當時咱們指望的是t1 t2兩個線程使some_var + 2變成17
在這裏就有了資源競爭。
相同的狀況也可能發生在其它的線程間，因此出現了最後的結果小於50的狀況。 app

二、解決資源競爭

from threading import Lock, Thread
lock = Lock()
some_var = 0
 
class IncrementThread(Thread):
    def run(self):
        #we want to read a global variable
        #and then increment it
        global some_var
        lock.acquire()
        read_value = some_var
        print "some_var in %s is %d" % (self.name, read_value)
        some_var = read_value + 1
        print "some_var in %s after increment is %d" % (self.name, some_var)
        lock.release()
 
def use_increment_thread():
    threads = []
    for i in range(50):
        t = IncrementThread()
        threads.append(t)
        t.start()
    for t in threads:
        t.join()
    print "After 50 modifications, some_var should have become 50"
    print "After 50 modifications, some_var is %d" % (some_var,)
 
use_increment_thread()

再次運行這個程序，達到了咱們預期的結果。
解釋：
Lock 用來防止競爭條件
若是在執行一些操做以前，線程t1得到了鎖。其餘的線程在t1釋放Lock以前，不會執行相同的操做
咱們想要肯定的是一旦線程t1已經讀取了some_var，直到t1完成了修改some_var，其餘的線程才能夠讀取some_var
這樣讀取和修改some_var成了邏輯上的原子操做。

示例3，多線程環境下的原子操做

讓咱們用一個例子來證實一個線程不能影響其餘線程內的變量（非全局變量）。
time.sleep()可使一個線程掛起，強制線程切換髮生。

一、BUG 版

from threading import Thread
import time
 
class CreateListThread(Thread):
    def run(self):
        self.entries = []
        for i in range(10):
            time.sleep(0.01)
            self.entries.append(i)
        print self.entries
 
def use_create_list_thread():
    for i in range(3):
        t = CreateListThread()
        t.start()
 
use_create_list_thread()

運行幾回後發現並無打印出正確的結果。當一個線程正在打印的時候，cpu切換到了另外一個線程，因此產生了不正確的結果。咱們須要確保print self.entries是個邏輯上的原子操做，以防打印時被其餘線程打斷。

二、加鎖保證操做的原子性

咱們使用了Lock()，來看下邊的例子。

from threading import Thread, Lock
import time
 
lock = Lock()
 
class CreateListThread(Thread):
    def run(self):
        self.entries = []
        for i in range(10):
            time.sleep(0.01)
            self.entries.append(i)
        lock.acquire()
        print self.entries
        lock.release()
 
def use_create_list_thread():
    for i in range(3):
        t = CreateListThread()
        t.start()
 
use_create_list_thread()

此次咱們看到了正確的結果。證實了一個線程不能夠修改其餘線程內部的變量（非全局變量）。

示例4，Python多線程簡易版：線程池 threadpool

上面的多線程代碼看起來有點繁瑣，下面咱們用 treadpool 將案例 1 改寫下：

import threadpool
import time
import urllib2

urls = [
    'http://www.google.com', 
    'http://www.amazon.com', 
    'http://www.ebay.com', 
    'http://www.alibaba.com', 
    'http://www.reddit.com'
]

def myRequest(url):
    resp = urllib2.urlopen(url)
    print url, resp.getcode()


def timeCost(request, n):
  print "Elapsed time: %s" % (time.time()-start)

start = time.time()
pool = threadpool.ThreadPool(5)
reqs = threadpool.makeRequests(myRequest, urls, timeCost)
[ pool.putRequest(req) for req in reqs ]
pool.wait()

解釋關鍵代碼：