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()查看线程的返回结果。
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为一个Tuture类对象, submit方法是非阻塞的,立即返回的。第二个参数为函数参数
tesk2 = excutor.submit(get_html, 2) print(task1.done()) # 判断函数是否执行成功 输出结果:
False
get page2 success
get page3 success
分析:因为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
主线程结束