집 > 기사 > 백엔드 개발 > Python의 멀티스레딩에 대한 자세한 소개(코드 예)

Python의 멀티스레딩에 대한 자세한 소개(코드 예)

不言원래의: 2018-08-29 10:18:461382검색

이 글은 Python의 멀티스레딩에 대한 자세한 소개(코드 예제)를 제공합니다. 필요한 친구들이 참고할 수 있기를 바랍니다.

이 문서에서는 Python을 학습할 때 발생하는 문제와 몇 가지 일반적인 사용법을 기록합니다. 이 개발 환경의 Python 버전은 2.7입니다.

1. Python 파일 이름 지정

Python 파일 이름을 지정할 때 시스템의 기본 모듈 이름과 충돌하지 않도록 주의해야 합니다. 그렇지 않으면 오류가 보고됩니다.
다음 예시와 같이 스레드 학습시 파일명이 threading.py이고, Python 스크립트는 완전히 정상인데, 다음과 같은 오류가 보고됩니다: AttributeError: '모듈' 개체에는 'xxx' 속성이 없습니다. threading.py,Python脚本完全正常没问题，结果报下面的错误：AttributeError: 'module' object has 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

问题定位：

查看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

查了下原因，是因为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 version portable imports would be to 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)

threading.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)

실행: #🎜🎜#

1999998000000
normal:  0.10034608840942383
1999998000000
multithreading:  0.08421492576599121

문제 위치:

#🎜🎜#Viewimport 라이브러리의 소스 파일이 존재하며 오류가 없는 것으로 확인되었습니다. 동시에 소스 파일의 .pyc 파일도 존재했습니다 #🎜🎜#

문제 해결 방법:

#🎜🎜#1.py 스크립트 이름을 지정할 때 Python 예약어, 모듈과 동일하게 이름을 지정하지 마세요. 이름 등.#🎜🎜#

#🎜🎜#2. 라이브러리의 .pyc 파일을 삭제합니다(py 스크립트가 .pyc를 생성하므로). .pyc 파일이 생성된 경우 코드가 업데이트되지 않은 경우 실행 시 pyc가 계속 사용되므로 .pyc 파일을 삭제하고 코드를 다시 실행하거나 환경을 찾으세요. 코드를 실행할 수 있는 코드를 실행하려면 현재 컴퓨터의 .pyc 파일을 복사하여 교체하세요#🎜🎜#

#🎜🎜#스크립트 파일 이름을 threading_test.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()

#🎜🎜#2. 멀티스레딩#🎜🎜##🎜🎜#멀티스레딩은 프로그램 계산 속도를 높이는 효과적인 방법입니다. Python의 멀티스레딩 모듈은 스레딩입니다. 빠르고 쉽게 시작할 수 있습니다. 이 섹션의 시작 부분에서 사용 방법을 알려드리겠습니다. #🎜🎜#

1. 스레드 추가

#🎜🎜#threading_test.py#🎜🎜#

➜  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)

2. 추가되지 않음 Join()의 결과#🎜🎜#T1 스레드 작업에 더 많은 시간이 소요됩니다#🎜🎜##🎜🎜#threading_join.py# 🎜🎜 #

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()

#🎜🎜#그러나 실제 실행 결과는: #🎜🎜#

# -*- 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()

join()의 결과 h4>#🎜🎜#스레드 작업이 완료되기 전에 all done을 출력합니다. 순서를 따르려면 스레드를 시작한 후 스레드에서 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)

#🎜🎜#결과 인쇄: #🎜🎜#rrreee#🎜🎜#전체 스크립트 파일 : #🎜🎜 #rrreee

해보기

#🎜🎜#두 개의 스레드를 추가하면 어떤 결과가 인쇄되나요? #🎜🎜#rrreee#🎜🎜#"one" 결과 출력은 다음과 같습니다. #🎜🎜#rrreee#🎜🎜#이제 T1이나 T2 모두 join이 없습니다. 모든 작업 수행 여부는 전적으로 두 스레드의 실행 속도에 따라 달라지므로 모든 작업이 완료된 후에 T2 종료가 나타날 수도 있습니다. 이 지저분한 실행 방법은 우리가 용납할 수 없으므로 조인을 사용하여 이를 제어해야 합니다. #🎜🎜##🎜🎜# T1이 시작된 후 T2가 시작되기 전에 thread_1.join()을 추가해 보겠습니다. #🎜🎜#rrreee#🎜🎜#결과 인쇄: #🎜🎜# rrreee#🎜 🎜#T2는 T1이 끝날 때까지 기다렸다가 실행을 시작하는 것을 볼 수 있습니다. #🎜🎜#

3. 처리 결과를 저장하는 큐

구현 함수

#🎜🎜#코드는 함수를 구현하고 데이터 목록의 데이터를 전송하며 4개의 스레드를 사용하여 결과는 Queue에 저장됩니다. 스레드가 실행된 후 저장된 결과는 Queue #🎜🎜##🎜🎜#멀티 스레드 함수에서 Queue를 정의합니다. 반환 값을 return 대신 저장하고 다중 스레드 목록을 정의하고 처리를 위해 다차원 데이터 목록을 초기화합니다. #🎜🎜##🎜🎜#threading_queue.py#🎜🎜#rrreee#🎜🎜 #위 스크립트 실행 시 오류가 발생했습니다: #🎜🎜#rrreee#🎜🎜# 원인을 확인한 결과 Python 버전으로 인해 발생했습니다: #🎜🎜#해결 방법: 이름이 지정된 모듈이 없습니다. 'Queue'#🎜🎜## 🎜🎜#Python 2에서는 모듈 이름이 Queue이고, Python 3에서는 PEP8 지침(모듈 이름은 모두 소문자)을 따르도록 이름이 바뀌어 클래스가 모두 Queue로 유지됩니다. 버전(PEP8 다음).#🎜 🎜##🎜🎜#일반적으로 버전 휴대용 가져오기를 작성하는 방식은 다음과 같습니다.#🎜🎜##🎜🎜#python3에서는 다음과 같이 인용할 수 있습니다. #🎜🎜# rrreee#🎜🎜#python2에서는 다음을 수행할 수 있습니다. 인용문: #🎜🎜#rrreee#🎜🎜#인쇄: #🎜🎜#rrreee#🎜🎜#전체 코드: #🎜🎜#threading_queue.py #🎜🎜#rrreee

4, GIL 효율성 문제

GIL이란 무엇인가요?

#🎜🎜#이번에는 Python의 멀티스레딩이 때때로 특별히 이상적이지 않은 이유를 살펴보겠습니다. 주된 이유는 Python의 설계에 필요한 링크, 즉 Global Interpreter가 있기 때문입니다. 잠금(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，最后打印输出的结果是不同的。