python中多線程的詳細介紹（程式碼範例）-Python教學-PHP中文網

首頁

後端開發

Python教學

python中多線程的詳細介紹（程式碼範例）

不言

Aug 29, 2018 am 10:18 AM

python

這篇文章帶給大家的內容是關於python中多線程的詳細介紹（程式碼範例），有一定的參考價值，有需要的朋友可以參考一下，希望對你有幫助。

本文記錄學習Python遇到的問題和一些常用用法，註本開發環境的Python版本為2.7。

一、python檔案命名

在python檔案命名時，一定要注意不能和系統預設的模組名稱衝突，否則會報錯。
如下面的例子，在學習線程時，將文件名命名為threading.py,Python腳本完全正常沒問題，結果報下面的錯誤：AttributeError: 'module' object有 no attribute 'xxx'。

threading.py

# -*- coding:utf-8 -*-

"""
@author: Corwien
@file: threading_test.py
@time: 18/8/25 09:14
"""

import threading

# 获取已激活的线程数
print(threading.active_count())

執行：

➜  baseLearn python threading/threading.py
Traceback (most recent call last):
  File "threading/threading.py", line 9, in <module>
    import threading
  File "/Users/kaiyiwang/Code/python/baseLearn/threading/threading.py", line 12, in <module>
    print(threading.active_count())
AttributeError: 'module' object has no attribute 'active_count'
➜  baseLearn</module></module>

問題定位：

查看import庫的來源文件，發現來源文件存在且沒有錯誤，同時存在來源文件的.pyc文件

#問題解決：

1.命名py腳本時，不要與python預留字，模組名稱等相同
#2.刪除該庫的.pyc」檔案（因為py腳本每次執行時均會產生.pyc檔；在已經產生.pyc檔的情況下，若程式碼不更新，執行時間依舊會走pyc，所以要刪除.pyc檔），重新執行程式碼；或找一個可以執行程式碼的環境，拷貝取代目前機器的.pyc檔即可

將腳本檔名重新命名為threading_test.py，然後執行，就不會報錯了。

➜  baseLearn python threading/threading_test.py
1
➜  baseLearn

二、多執行緒threading

多執行緒是加速程式運算的有效方式，Python的多執行緒模組threading上手快速簡單，從這節開始我們就教大家如何使用它。

1、新增執行緒

threading_test.py

# -*- coding:utf-8 -*-

"""
@author: Corwien
@file: threading_test.py
@time: 18/8/25 09:14
"""

import threading

# 获取已激活的线程数
# print(threading.active_count())

# 查看所有线程信息
# print(threading.enumerate())

# 查看现在正在运行的线程
# print(threading.current_thread())

def thread_job():
    print('This is a thread of %s' % threading.current_thread())


def main():
    thread = threading.Thread(target=thread_job,)  # 定义线程
    thread.start() # 让线程开始工作

if __name__ == '__main__':
    main()

2、join功能

#不加join() 的結果

我們讓T1 執行緒工作的耗時增加

threading_join.py

# -*- coding:utf-8 -*-

"""
@author: Corwien
@file: threading_join.py
@time: 18/8/25 09:14
"""

import threading
import time

def thread_job():
    print('T1 start\n')
    for i in range(10):
        time.sleep(0.1) # 任务时间0.1s
    print("T1 finish\n")


def main():
    added_thread = threading.Thread(target=thread_job, name='T1')  # 定义线程
    added_thread.start() # 让线程开始工作
    print("all done\n")

if __name__ == '__main__':
    main()

預想中輸出的結果是按照順序依序往下執行：

T1 start
T1 finish
all done

但實際執行結果為：

➜  baseLearn python threading/threading_join.py
T1 start
all done


T1 finish

➜  baseLearn

加入join()的結果

執行緒任務尚未完成便輸出all done。如果要遵循順序，可以在啟動執行緒後對它呼叫join：

added_thread.start()
added_thread.join()
print("all done\n")

列印結果：

➜  baseLearn python threading/threading_join.py
T1 start

T1 finish

all done

完整腳本檔案：

# -*- coding:utf-8 -*-

"""
@author: Corwien
@file: threading_join.py
@time: 18/8/25 09:14
"""

import threading
import time

def thread_job():
    print('T1 start\n')
    for i in range(10):
        time.sleep(0.1) # 任务时间0.1s
    print("T1 finish\n")


def main():
    added_thread = threading.Thread(target=thread_job, name='T1')  # 定义线程
    added_thread.start() # 让线程开始工作
    added_thread.join()
    print("all done\n")

if __name__ == '__main__':
    main()

小試牛刀

如果加入兩個線程，列印的輸出結果是怎麼樣的呢？

# -*- coding:utf-8 -*-

"""
@author: Corwien
@file: threading_join.py
@time: 18/8/25 09:14
"""

import threading
import time

def T1_job():
    print('T1 start\n')
    for i in range(10):
        time.sleep(0.1) # 任务时间0.1s
    print("T1 finish\n")

def T2_job():
    print("T2 start\n")
    print("T2 finish\n")

def main():
    thread_1 = threading.Thread(target=T1_job, name='T1')  # 定义线程
    thread_2 = threading.Thread(target=T2_job, name='T2')  # 定义线程
    thread_1.start()  # 开启T1
    thread_2.start()  # 开启T2
    print("all done\n")

if __name__ == '__main__':
    main()

輸出的」一種」結果是：

T1 start

T2 start

T2 finish

all done

T1 finish

現在T1和T2都沒有join，注意這裡說」一種」是因為all done的出現完全取決於兩個執行緒的執行速度，完全有可能T2 finish出現在all done之後。這種雜亂的執行方式是我們不能忍受的，因此要使用join加以控制。

我們試試在T1啟動後，T2啟動前加上thread_1.join():

thread_1.start()
thread_1.join() # notice the difference!
thread_2.start()
print("all done\n")

列印結果：

T1 start

T1 finish

T2 start
all done

T2 finish

可以看到，T2會等待T1結束後才開始運作。

3、儲存進程結果Queue

實作功能

程式碼實作功能，將資料清單中的資料傳入，使用四個執行緒處理，將結果保存在Queue中，執行緒執行完後，從Queue中取得儲存的結果

在多執行緒函數中定義一個Queue，用來儲存回傳值，取代return，定義一個多線程列表，初始化一個多維資料列表，用來處理：

threading_queue.py

# -*- coding:utf-8 -*-

"""
@author: Corwien
@file: threading_queue.py
@time: 18/8/25 09:14
"""

import threading
import time
from queue import Queue

def job(l, q):
    for i in range(len(l)):
        l[i] = l[i] ** 2
    q.put(l) #多线程调用的函数不能用return返回值

def multithreading():
    q = Queue()  #q中存放返回值，代替return的返回值
    threads = []
    data = [[1,2,3],[3,4,5],[4,4,4],[5,5,5]]

    for i in range(4): #定义四个线程
        t = threading.Thread(target=job, args=(data[i], q))  #Thread首字母要大写，被调用的job函数没有括号，只是一个索引，参数在后面
        t.start() #开始线程
        threads.append(t) #把每个线程append到线程列表中

    for thread in threads:
        thread.join()

    results = []
    for _ in range(4):
        results.append(q.get()) #q.get()按顺序从q中拿出一个值
    print(results)


if __name__ == '__main__':
    multithreading()

執行上邊的腳本出現了這樣的錯誤：

➜  baseLearn python threading/threading_queue.py
Traceback (most recent call last):
  File "threading/threading_queue.py", line 11, in <module>
    from queue import Queue
ImportError: No module named queue</module>

查了下原因，是因為python版本導致的:
解決方法：No module named 'Queue'

On Python 2, the module is named Queue, on Python 3, it was renamed to follow PEP8 guidelines (all lowercase for module names), making it queue. The class remains Queue on all versions (following PEP8).

Typically, the way you'd write portion portable w portion port be imbe todable imbe 到do:

python3 中這樣引用：

try:
    import queue
except ImportError:
    import Queue as queue

在python2 中我們可以這樣引用：

from Queue import Queue

列印：

baseLearn python ./threading/threading_queue.py
[[1, 4, 9], [9, 16, 25], [16, 16, 16], [25, 25, 25]]

完整程式碼：
threading_queue.py

# -*- coding:utf-8 -*-

"""
@author: Corwien
@file: threading_queue.py
@time: 18/8/25 09:14
"""

import threading
# import time
from Queue import Queue

def job(l, q):
    for i in range(len(l)):
        l[i] = l[i] ** 2
    q.put(l) #多线程调用的函数不能用return返回值

def multithreading():
    q = Queue()  #q中存放返回值，代替return的返回值
    threads = []
    data = [[1,2,3],[3,4,5],[4,4,4],[5,5,5]]

    for i in range(4): #定义四个线程
        t = threading.Thread(target=job, args=(data[i], q))  #Thread首字母要大写，被调用的job函数没有括号，只是一个索引，参数在后面
        t.start() #开始线程
        threads.append(t) #把每个线程append到线程列表中

    for thread in threads:
        thread.join()

    results = []
    for _ in range(4):
        results.append(q.get()) #q.get()按顺序从q中拿出一个值
    print(results)


if __name__ == '__main__':
    multithreading()

4、GIL效率問題

何為GIL？

這次我們來看看為什麼說python 的多線程threading 有時候並不是特別理想. 最主要的原因是就是, Python 的設計上, 有一個必要的環節, 就是Global Interpreter Lock (GIL)。這個東西讓 Python 還是一次只能處理一個東西。

GIL的解釋：

尽管Python完全支持多线程编程，但是解释器的C语言实现部分在完全并行执行时并不是线程安全的。实际上，解释器被一个全局解释器锁保护着，它确保任何时候都只有一个Python线程执行。 GIL最大的问题就是Python的多线程程序并不能利用多核CPU的优势（比如一个使用了多个线程的计算密集型程序只会在一个单CPU上面运行）。在讨论普通的GIL之前，有一点要强调的是GIL只会影响到那些严重依赖CPU的程序（比如计算型的）。如果你的程序大部分只会涉及到I/O，比如网络交互，那么使用多线程就很合适，因为它们大部分时间都在等待。实际上，你完全可以放心的创建几千个Python线程，现代操作系统运行这么多线程没有任何压力，没啥可担心的。

测试GIL

我们创建一个 job, 分别用 threading 和一般的方式执行这段程序. 并且创建一个 list 来存放我们要处理的数据. 在 Normal 的时候, 我们这个 list 扩展4倍, 在 threading 的时候, 我们建立4个线程, 并对运行时间进行对比.

threading_gil.py

# -*- coding:utf-8 -*-

"""
@author: Corwien
@file: threading_gil.py
@time: 18/8/25 09:14
"""

import threading
from Queue import Queue
import copy
import time

def job(l, q):
    res = sum(l)
    q.put(l) #多线程调用的函数不能用return返回值

def multithreading(l):
    q = Queue()  #q中存放返回值，代替return的返回值
    threads = []

    for i in range(4): #定义四个线程
        t = threading.Thread(target=job, args=(copy.copy(l), q), name="T%i" % i)  #Thread首字母要大写，被调用的job函数没有括号，只是一个索引，参数在后面
        t.start() #开始线程
        threads.append(t) #把每个线程append到线程列表中

    [t.join() for t in threads]
    total = 0
    for _ in range(4):
        total = q.get() #q.get()按顺序从q中拿出一个值
    print(total)

def normal(l):
    total = sum(l)
    print(total)

if __name__ == '__main__':
    l = list(range(1000000))
    s_t = time.time()
    normal(l*4)
    print('normal:', time.time() - s_t)
    s_t = time.time()
    multithreading(l)
    print('multithreading: ', time.time() - s_t)

如果你成功运行整套程序, 你大概会有这样的输出. 我们的运算结果没错, 所以程序 threading 和 Normal 运行了一样多次的运算. 但是我们发现 threading 却没有快多少, 按理来说, 我们预期会要快3-4倍, 因为有建立4个线程, 但是并没有. 这就是其中的 GIL 在作怪.

1999998000000
normal:  0.10034608840942383
1999998000000
multithreading:  0.08421492576599121

5、线程锁Lock

不使用 Lock 的情况

threading_lock.py

# -*- coding:utf-8 -*-

"""
@author: Corwien
@file: threading_lock.py
@time: 18/8/25 09:14
"""

import threading

# 全局变量A的值每次加1，循环10次，并打印
def job1():
    global A
    for i in range(10):
        A+=1
        print('job1',A)

# 全局变量A的值每次加10，循环10次，并打印
def job2():
    global A
    for i in range(10):
        A+=10
        print('job2',A)

# 定义两个线程，分别执行函数一和函数二
if __name__== '__main__':
   
    A=0
    t1=threading.Thread(target=job1)
    t2=threading.Thread(target=job2)
    t1.start()
    t2.start()
    t1.join()
    t2.join()

打印输出数据：

➜  baseLearn python ./threading/threading_lock.py
('job1', ('job2'1)
, (11)'job1'
('job2', 22)
('job2', 32)
('job2', 42)
('job2', 52)
('job2', 62)
('job2', 72)
('job2', 82)
('job2', 92)
('job2', 102)
, 12)
('job1', 103)
('job1', 104)
('job1', 105)
('job1', 106)
('job1', 107)
('job1', 108)
('job1', 109)
('job1', 110)

可以看出，打印的结果非常混乱

使用 Lock 的情况

lock在不同线程使用同一共享内存时，能够确保线程之间互不影响，使用lock的方法是，在每个线程执行运算修改共享内存之前，执行lock.acquire()将共享内存上锁，确保当前线程执行时，内存不会被其他线程访问，执行运算完毕后，使用lock.release()将锁打开，保证其他的线程可以使用该共享内存。

函数一和函数二加锁

def job1():
    global A,lock
    lock.acquire()
    for i in range(10):
        A+=1
        print('job1',A)
    lock.release()

def job2():
    global A,lock
    lock.acquire()
    for i in range(10):
        A+=10
        print('job2',A)
    lock.release()

主函数中定义一个Lock

if __name__== '__main__':
    lock=threading.Lock()
    A=0
    t1=threading.Thread(target=job1)
    t2=threading.Thread(target=job2)
    t1.start()
    t2.start()
    t1.join()
    t2.join()

完整代码：

# -*- coding:utf-8 -*-

"""
@author: Corwien
@file: threading_lock.py
@time: 18/8/25 09:14
"""

import threading

def job1():
    global A,lock
    lock.acquire()
    for i in range(10):
        A+=1
        print('job1',A)
    lock.release()

def job2():
    global A,lock
    lock.acquire()
    for i in range(10):
        A+=10
        print('job2',A)
    lock.release()

if __name__== '__main__':
    lock = threading.Lock()
    A=0
    t1=threading.Thread(target=job1)
    t2=threading.Thread(target=job2)
    t1.start()
    t2.start()
    t1.join()
    t2.join()

打印输出：

➜  baseLearn python ./threading/threading_lock.py
('job1', 1)
('job1', 2)
('job1', 3)
('job1', 4)
('job1', 5)
('job1', 6)
('job1', 7)
('job1', 8)
('job1', 9)
('job1', 10)
('job2', 20)
('job2', 30)
('job2', 40)
('job2', 50)
('job2', 60)
('job2', 70)
('job2', 80)
('job2', 90)
('job2', 100)
('job2', 110)

从打印结果来看，使用lock后，一个一个线程执行完。使用lock和不使用lock，最后打印输出的结果是不同的。