Python的GIL机制与多线程编程

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

GIL

全称global interpreter lock 全局解释锁

gil使得python同一个时刻只有一个线程在一个cpu上执行字节码，并且无法将多个线程映射到多个cpu上，即不能发挥多个cpu的优势。

gil会根据执行的字节码行数以及时间片释放gil，也会在遇到IO操作时候主动释放。

线程

操作系统能够调动的最小单元就是线程。最开始是进程，因为进程对资源的消耗大，所以演变成了线程。

对于IO操作来说，多线程和多进程性能差别不大。

方式一：通过thread类实例化

import threading
import time
def get_html(url):
print('get html started')
time.sleep(2)
print('get html ended')
def get_url(url):
print('get url started')
time.sleep(2)
print('get url ended')get_html = threading.Thread(target=get_html, args=('url1',))
get_url = threading.Thread(target=get_url, args=('url2',))if __name__ =='__main__':
start_time = time.time()
get_html.start()
get_url.start()
print(time.time() - start_time)
输出结果：
get html started
get url started
0.0009999275207519531
get html ended
get url ended

此处因为自定义了两个线程，但是实际有三个线程，（还有一个主线程）因为直接线程.start()是非阻塞的，所以先会运行打印时间，然后再结束上面两个线程。如果想要等上面两个线程结束之后再执行主线程打印出时间话(即阻塞)可以有两种方法

①在线程开始前加入语言：(只要主线程结束之后就结束整个程序，Kill所有的子线程)

get_html.setDaemon(True)

get_url.setDaemon(True)

②在线程开始之后加入语言（将等待线程运行结束之后再往下继续执行代码）：

get_html.join()

get_url.join()

方式二：继承threading.Thread类

import threading
import time
class GetHtml(threading.Thread):
def __init__(self, name):
super().__init__(name=name)
def run(self):
print('get html started')
time.sleep(2)
print('get html ended')class GetUrl(threading.Thread):
def __init__(self, name):
super().__init__(name=name)
def run(self):
print('get url started')
time.sleep(2)
print('get url ended')get_html = GetHtml('HTML')
get_url = GetUrl('URL')if __name__ =='__main__':
start_time =time.time()
get_html.start()
get_url.start()
get_html.join()
get_url.join()
print(time.time() - start_time)
输出结果：
get html started
get url started
get html ended
get url ended
2.0011143684387207

线程间的通信

1 通过全局变量进行通信

import time
import threading
url_list = []
def get_html():
global url_list
url = url_list.pop()
print('get html form {} started'.format(url))
time.sleep()
print('get html from {} ended'.format(url))def get_url():
global url_list
print('get url started')
time.sleep()
for i in range():
url_list.append('http://www.baidu.com/{id}'.format(id=i))
print('get url ended')
if __name__ == '__main__':
thread_url = threading.Thread(target=get_url)
for i in range():
thread_html = threading.Thread(target=get_html)
thread_html.start()

上述代码比较原始，不灵活，可以将全局变量url_list通过参数传入函数调用

import time
import threading
url_list = []
def get_html(url_list):
url = url_list.pop()
print('get html form {} started'.format(url))
time.sleep(1)
print('get html from {} ended'.format(url))
def add_url(url_list):
print('add url started')
time.sleep(1)
for i in range(20):
url_list.append('http://www.baidu.com/{id}'.format(id=i))
print('add url ended')
if __name__ == '__main__':
thread_url = threading.Thread(target=add_url, args=(url_list,))
thread_url.start()
thread_url.join()
for i in range(20):
thread_html = threading.Thread(target=get_html, args=(url_list,))
thread_html.start()

还有一种方式为新建一个py文件，然后在文件中定义一个变量，url_list = [] 然后开头的时候用import导入这个变量即可。这种方式对于变量很多的情况下为避免混乱统一将变量进行管理。但是此方式一定要注意import的时候只要import到文件，而不要import到变量。（比如说文件名为variables.python内定义一个变量名url_list=[]，需要import variables，然后代码中用variables.url_list 而不是 from variables import url_list 因为后一种方式导入的话，在其他线程修改此变量的时候，我们是看不到的。但是第一种方式可以看到。

总结：不管以何种形式共享全局变量，都不是线程安全的操作，所以为了达到线程安全，就需要用到线程锁，lock的机制，代码就会比较复杂，所有引入了一种安全的线程通信，from queue import Queue

2用消息队列Queue（推荐使用，Queue是线程安全的，不会冲突的）

import time
import threading
from queue import Queue
def get_html(queue):
url = queue.get()
print('get html form {} started'.format(url))
time.sleep(1)
print('get html from {} ended'.format(url))def add_url(queue):
print('add url started')
time.sleep(1)
for i in range(20):
queue.put('http://www.baidu.com/{id}'.format(id=i))
print('add url ended')
if __name__ == '__main__':
url_queue = Queue(maxsize=1000) # 设置队列中元素的max个数。
thread_url = threading.Thread(target=add_url, args=(url_queue,))
thread_url.start()
thread_url.join()
list1=[]
for i in range(20):
thread_html = threading.Thread(target=get_html, args=(url_queue,))
list1.append(thread_html)
for i in list1:
i.start()

线程同步的问题：

概念：

线程的同步（即当有一个线程在对内存进行操作时，其他线程都不可以对这个内存地址进行操作，直到该线程完成操作，其他线程才能对该内存地址进行操作，而其他线程又处于等待状态）

线程为什么要同步？

问题：既然python有GIL机制，那么线程就是安全的，那么为什么还有线程同步问题？

回到上面GIL的介绍（gil会根据执行的字节码行数以及时间片释放gil，也会在遇到IO操作时候主动释放）

再看一个经典的案列：如果GIL使线程绝对安全的话，那么最后结果恒为0，事实却不是这样。

from threading import Thread
total = 0
def add():
global total
for i in range(1000000):
total += 1
def desc():
global total
for i in range(1000000):
total -= 1
thread1 = Thread(target=add)
thread2 = Thread(target=desc)
thread1.start()
thread2.start()
thread1.join()
thread2.join()
print(total)312064

结果打印不稳定，都不会0，

线程同步的方法：

1：线程锁机制 Lock

注意，锁的获取和释放也需要时间，于是会对程序的运行性能产生一定的影响。而且极易造成死锁，于是对应的可以将Lock改为Rlock，就可以支持同时多个acquire进入锁，但是一定注意，Rlock只在单线程内起作用，并且acquire次数要和release次数想等。

import threading
from threading import Lock
l = Lock()
a = 0
def add():
global a
global l
l.acquire()
for i in range(1000000):
a += i
l.release() # 记得线程段结束运行之后一定需要解锁。不然其他程序就阻塞了。
def desc():
global a
global l
l.acquire()
for i in range(1000000):
a -= i
l.release()
thread1 = threading.Thread(target=add)
thread2 = threading.Thread(target=desc)
thread1.start()
thread2.start()
thread1.join() # 再次注意如果线程只是start()没有join()的话，那么任意线程执行完了就会往下执行print语句，但是如果加了join的话，就会等thread1和thread2运行完之后在运行下面的语句。
thread2.join()
print(a)输出结果恒为0

2：条件变量：condition

复杂的线程通讯的话lock机制已经不再适用，例如：

from threading import Condition, Thread, Lock
# 条件变量，用复杂的线程间的同步
lock = Lock()class Tom(Thread):
def __init__(self, lock):
self.lock = lock
super().__init__(name='Tom') def run(self):
self.lock.acquire()
print('{}: hello, Bob.'.format(self.name))
self.lock.release()
self.lock.acquire()
print("{}: Let's have a chat.".format(self.name))
self.lock.release()class Bob(Thread):
def __init__(self, lock):
self.lock = lock
super().__init__(name='Bob') def run(self):
self.lock.acquire()
print('{}: Hi, Tom.'.format(self.name))
self.lock.release()
self.lock.acquire()
print("{}:Well, I like to talk to you.".format(self.name))
self.lock.release()tom = Tom(lock)
bob = Bob(lock)
tom.start()
bob.start()Tom: hello, Bob.
Tom: Let's have a chat.
Bob: Hi, Tom.
Bob:Well, I like to talk to you.

为什么会这样？原因很简单，Tom在start()的时候，还没有来得及Bob start()之前就将所有的逻辑执行完了，其次，GIL切换的时候是根据时间片或者字节码行数来的，即也可能因为在时间片内将Tom执行完毕之后才切换到Bob。于是引入了条件变量机制，condition，

看condition原代码可以了解到，其集成了魔法方法__enter__ 和 __exit__于是可以用with语句调用，在__enter__方法中，调用了

def __enter__(self):
return self._lock.__enter__()

而__enter__() 方法则直接调用了acquire方法，同时acquire其实就是调用了Rlock.acquire()方法。所以condition内部其实还是使用了Rlock方法来实现。同理__exit__则调用了Rlock.release()

重要方法 wait()和notify()

wait()允许我们等待某个条件变量的通知，而notify()方法则是发送一个通知。于是就可以修改上述代码：

from threading import Condition, Thread, Lock
# 条件变量，用复杂的线程间的同步class Tom(Thread):
def __init__(self, condition):
self.condition = condition
super().__init__(name='Tom') def run(self):
with self.condition:
print('{}: hello, Bob.'.format(self.name))
self.condition.notify()
self.condition.wait()
print("{}: Let's have a chat.".format(self.name))
self.condition.notify()class Bob(Thread):
def __init__(self, condition):
self.condition = condition
super().__init__(name='Bob') def run(self):
with self.condition:
self.condition.wait()
print('{}: Hi, Tom.'.format(self.name))
self.condition.notify()
self.condition.wait()
print("{}:Well, I like to talk to you.".format(self.name))if __name__ == '__main__':
condition = Condition()
tom = Tom(condition)
bob = Bob(condition) bob.start()
tom.start()

上述代码注意：

开始顺序很重要，因为wait()方法必须要notify()方法才能唤醒，如果先调用tom.start()的话，那么当tom中的self.condition.notify()调用完毕之后，bob开没有开始启动，所以根本接受不到tom的信号，于是要先调用bob的wait()使其处于一个类似监听状态。
必须要使用with self.condition，或者是self.condition.acquire()之后才能使用后面的wait()和notify()方法。
如果上面不是用with方法打开的self.condition那么在代码结束之后一定要记得self.condition.release()释放锁。

3：semaphore

用于控制进入某段代码线程的数量，比如说做爬虫的时候，在请求页面的时候防止线程数量过多，短时间内请求频繁被发现，可以使用semaphore来控制进入请求的线程数量。

from threading import Thread, Semaphore, Condition, Lock, RLock
import time
class GetHtml(Thread):
def __init__(self, url, sem):
super().__init__()
self.url = url
self.sem = sem
def run(self):
time.sleep(2)
print('get html successful.')
self.sem.release() # 开启之后记得要释放。
class GetUrl(Thread):
def __init__(self, sem):
super().__init__()
self.sem = sem
def run(self):
for i in range(20):
self.sem.acquire() # 开启semaphore
get_html = GetHtml('www.baidu.com/{}'.format(i), self.sem)
get_html.start()
if __name__ == '__main__':
sem = Semaphore(3) # 接受一个参数，设置最大进入的线程数为3
get_url = GetUrl(sem)
get_url.start()

线程池（比semaphore更加容易实现线程数量的控制）

from concurrent import futures

出了控制线程数量的其它功能：

主线程可以获取某一个线程的状态，以及返回值。
当一个线程我完成的时候，我们可以立即知道。
futures可以让多线程可多进程的编码接口一致。多进程改多线程或者多线程改多进程代码的时候，切换会非常平滑。

注意下代码中的task1，task2都是线程池建立的一个Future对象，此对象的设计非常重要， Future可以看做是一个未来对象，或者说是一个线程的状态收集容器，可以通过它的.done()查看线程是否运行结束，也可以通过.result()查看线程的返回结果。

分析：因为submit方法是非阻塞的，立即返回的。后面的print代码不会等待task1运行结束。如果加入等待时间等待task1完成则将返回True：

import time
from concurrent.futures import ThreadPoolExecutor
def get_html(times):
time.sleep(times)
print('get page{} success'.format(times))
return times
excutor = ThreadPoolExecutor(max_workers=2)
task1 = excutor.submit(get_html, 3) #task1为一个futures类对象， submit方法是非阻塞的，立即返回的。第二个参数为函数参数
tesk2 = excutor.submit(get_html, 2)print(task1.done()) # 判断函数是否执行成功
time.sleep(4)
print(task1.done())
输出结果：
False
get page2 success
get page3 success
True

代码后面加入

print(task1.result()) # 用result()方法可以获取到线程函数返回的结果。

可以用result()方法可以获取到线程函数返回的结果。

用代码：print(task1.cancel())可以将task1在运行之前取消掉，如果取消成功则返回True，反之False

import time
from concurrent.futures import ThreadPoolExecutor
def get_html(times):
time.sleep(times)
print('get page{} success'.format(times))
return times
excutor = ThreadPoolExecutor(max_workers=1) # 将线程池数量改为1，让tesk2先等待不执行，方便取消。task1 = excutor.submit(get_html, 3) #task1为一个futures类对象， submit方法是非阻塞的，立即返回的。第二个参数为函数参数
tesk2 = excutor.submit(get_html, 2)print(task1.done()) # 判断函数是否执行成功
print(tesk2.cancel())
time.sleep(4)
print(task1.done())
print(task1.result()) # 用result()方法可以获取到线程函数返回的结果。输出结果：（结果无get page 2 sucess）
False
True
get page3 success
True
3

在某些情况下，要获取已经成功的task的返回值。

方法一：需要用到as_complete

import time
from concurrent.futures import ThreadPoolExecutor, as_completed
def get_html(times):
time.sleep(times)
print('get page{} success'.format(times))
return times
excutor = ThreadPoolExecutor(max_workers=2)
urls = [3, 2, 4]
all_task = [excutor.submit(get_html, url) for url in urls]
for futures in as_completed(all_task):
data = futures.result()
print('get {} page'.format(data))
输出结果：
get page2 success
get 2 page
get page3 success
get 3 page
get page4 success
get 4 page

代码分析：可以看到因为excutor.submit()是非阻塞的，由打印结果可以看出，没一个线程执行成功之后，as_complete()就会拿到其结果。

方法二：用executor.map

import time
from concurrent.futures import ThreadPoolExecutor, as_completed
def get_html(times):
time.sleep(times)
print('get page{} success'.format(times))
return times
excutor = ThreadPoolExecutor(max_workers=2)
urls = [3, 2, 4]
for data in excutor.map(get_html, urls):
print('get {} page'.format(data))结果：
get page2 success
get page3 success
get 3 page
get 2 page
get page4 success
get 4 page

可以看到用excutor.map方法不是完成一个打印一个，而是按照参数列表中的顺序，先get第一个参数结果，然后依次get，推荐可以使用第一种as_complete()方式。

wait方法使主线程阻塞

等待所有线程完成之后再往下走，wait()里面也可以选择参数return_when，默认是ALL_COMPLETE，如果为FIRST_COMPLETE(注意该参数需要在前面的import先导入)则第一个执行完成之后就会往下执行。

import time
from concurrent.futures import ThreadPoolExecutor, as_completed, wait
def get_html(times):
time.sleep(times)
print('get page{} success'.format(times))
return times
excutor = ThreadPoolExecutor(max_workers=2)
urls = [3, 2, 4]
all_task = [excutor.submit(get_html, url) for url in urls]
wait(all_task)
print('主线程结束')打印结果：get page2 success
get page3 success
get page4 success
主线程结束