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 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)。这个东西让 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，最后打印输出的结果是不同的。