GIL全局解释器锁
'''
python解释器:
- Cpython C语言
- Jpython java
... 1、GIL: 全局解释器锁
- 翻译: 在同一个进程下开启的多线程，同一时刻只能有一个线程执行，因为Cpython的内存管理不是线程安全。 - GIL全局解释器锁，本质上就是一把互斥锁，保证数据安全。 定义：
In CPython, the global interpreter lock, or GIL, is a mutex that prevents multiple
native threads from executing Python bytecodes at once. This lock is necessary mainly
because CPython’s memory management is not thread-safe. (However, since the GIL
exists, other features have grown to depend on the guarantees that it enforces.) 结论：在Cpython解释器中，同一个进程下开启的多线程，同一时刻只能有一个线程执行，无法利用多核优势。 # 为什么要有全局解释器锁：
- 没有锁: 2、GIL全局解释器锁的优缺点:
1.优点:
保证数据的安全
2.缺点:
单个进程下，开启多个线程，牺牲执行效率，无法实现并行，只能实现并发。 - IO密集型， 多线程
- 计算密集型，多进程
'''
import time
from threading import Thread n = 100 def task():
global n
m = n
time.sleep(3) #遇到IO,保存状态+切换，其他线程继续争抢GIL
n = m-1 if __name__ == '__main__':
list1 = []
for line in range(10):
t = Thread(target=task)
t.start()
list1.append(t) for t in list1:
t.join() print(n) #
多线程多进程效率比较
单核情况下都用多线程效率高
'''
IO密集型下使用多线程.
计算密集型下使用多进程. IO密集型任务，每个任务4s
- 单核:
- 开启线程比进程节省资源。 - 多核:
- 多线程:
- 开启4个子线程: 16s - 多进程:
- 开启4个进程: 16s + 申请开启资源消耗的时间 计算密集型任务，每个任务4s
- 单核:
- 开启线程比进程节省资源。 - 多核:
- 多线程:
- 开启4个子线程: 16s - 多进程:
- 开启多个进程: 4s 计算密集型: 多进程
假设100份原材料同时到达工厂，聘请100个人同时制造，效率最高 IO密集型： 多线程
假设100份原材料，只有40份了，其他还在路上，聘请40个人同时制造。 '''
from threading import Thread
from multiprocessing import Process
import time # 计算密集型任务
def task1():
# 计算1000000次 += 1
i = 10
for line in range(10000000):
i += 1 # IO密集型任务
def task2():
time.sleep(3) if __name__ == '__main__':
# 1、测试多进程:
# 测试计算密集型
# start_time = time.time()
# list1 = []
# for line in range(6):
# p = Process(target=task1)
# p.start()
# list1.append(p)
#
# for p in list1:
# p.join()
# end_time = time.time()
# # 消耗时间: 4.204240560531616
# print(f'计算密集型消耗时间: {end_time - start_time}')
#
# # 测试IO密集型
# start_time = time.time()
# list1 = []
# for line in range(6):
# p = Process(target=task2)
# p.start()
# list1.append(p)
#
# for p in list1:
# p.join()
# end_time = time.time()
# # 消耗时间: 4.382250785827637
# print(f'IO密集型消耗时间: {end_time - start_time}') # 2、测试多线程:
# 测试计算密集型
start_time = time.time()
list1 = []
for line in range(6):
p = Thread(target=task1)
p.start()
list1.append(p) for p in list1:
p.join()
end_time = time.time()
# 消耗时间: 5.737328052520752
print(f'计算密集型消耗时间: {end_time - start_time}') # 测试IO密集型
start_time = time.time()
list1 = []
for line in range(6):
p = Thread(target=task2)
p.start()
list1.append(p) for p in list1:
p.join()
end_time = time.time()
# 消耗时间: 3.004171848297119
print(f'IO密集型消耗时间: {end_time - start_time}')
进程池
# 线程池
from concurrent.futures import ProcessPoolExecutor
import time
# 池子对象: 内部可以帮你提交50个启动进程的任务
p_pool = ProcessPoolExecutor(50) def task1(n):
print(f'from task1...{n}')
time.sleep(10) if __name__ == '__main__':
n = 1
while True:
# 参数1: 函数名
# 参数2: 函数的参数1
# 参数3: 函数的参数2
# submit(参数1, 参数2, 参数3)
p_pool.submit(task1, n)
n += 1
wait方法可以让主线程阻塞，直到满足设定的要求。
from concurrent.futures import ThreadPoolExecutor, wait, ALL_COMPLETED, FIRST_COMPLETED
import time # 参数times用来模拟网络请求的时间
def get_html(times):
time.sleep(times)
print("get page {}s finished".format(times))
return times executor = ThreadPoolExecutor(max_workers=2)
urls = [3, 2, 4] # 并不是真的url
all_task = [executor.submit(get_html, (url)) for url in urls]
wait(all_task, return_when=ALL_COMPLETED)
print("main")
# 执行结果
# get page 2s finished
# get page 3s finished
# get page 4s finished
# main
wait方法接收3个参数，等待的任务序列、超时时间以及等待条件。等待条件return_when默认为ALL_COMPLETED，表明要等待所有的任务都结束。可以看到运行结果中，确实是所有任务都完成了，主线程才打印出main。等待条件还可以设置为FIRST_COMPLETED，表示第一个任务完成就停止等待。用线程池爬取梨视频
# 线程池测试
############
# 2 爬取视频
#############
# 先查看ajax的调用路径 https://www.pearvideo.com/category_loading.jsp?reqType=5&categoryId=9&start=0
#categoryId=9 分类id
#start=0 从哪个位置开始，每次加载12个
# https://www.pearvideo.com/category_loading.jsp?reqType=5&categoryId=9&start=0 from concurrent.futures import ThreadPoolExecutor,wait,ALL_COMPLETED
import requests
import re
import time
start = time.time() pool = ThreadPoolExecutor(10)
ret = requests.get('https://www.pearvideo.com/category_loading.jsp?reqType=5&categoryId=1&start=0')
# print(ret.text)
# 正则解析
reg = '<a href="(.*?)" class="vervideo-lilink actplay">'
video_urls=re.findall(reg,ret.text)
# print(video_urls) def download(url):
ret_detail = requests.get('https://www.pearvideo.com/' + url) # 相当于文件句柄，建立连接，流的方式，不是一次性取出
# print(ret_detail.text)
# 正则去解析
reg = 'srcUrl="(.*?)",vdoUrl=sr' # 正则表达式匹配式
mp4_url = re.findall(reg, ret_detail.text)[0] # type:str
# 下载视频
video_content = requests.get(mp4_url)
video_name = mp4_url.rsplit('/', 1)[-1] with open(video_name, 'wb')as f:
for line in video_content.iter_content():
f.write(line) threads=[]
for url in video_urls:
t = pool.submit(download(url),url)
threads.append(t) wait(threads, return_when=ALL_COMPLETED) # 等待线程池全部完成
end = time.time()
print('time:',end-start) # 计算时间
线程池其他方法可以参考： https://www.jianshu.com/p/b9b3d66aa0be
协程（理论 + 代码）
1.什么是协程？
- 进程: 资源单位
- 线程: 执行单位
- 协程: 单线程下实现并发
- 在IO密集型的情况下，使用协程能提高最高效率。
注意: 协程不是任何单位，只是一个程序员YY出来的东西。
Nignx
500 ----> 500 ---> 250000 ---> 250000 ----> 10 ----> 2500000
总结: 多进程 ---》 多线程 ---》 让每一个线程都实现协程.(单线程下实现并发)
- 协程的目的:
- 手动实现 "遇到IO切换 + 保存状态" 去欺骗操作系统，让操作系统误以为没有IO操作，将CPU的执行权限给你。
from gevent import monkey # 猴子补丁
monkey.patch_all() # 监听所有的任务是否有IO操作
from gevent import spawn # spawn(任务)
from gevent import joinall
import time def task1():
print('start from task1...')
time.sleep(1)
print('end from task1...') def task2():
print('start from task2...')
time.sleep(3)
print('end from task2...') def task3():
print('start from task3...')
time.sleep(5)
print('end from task3...') if __name__ == '__main__': start_time = time.time()
sp1 = spawn(task1)
sp2 = spawn(task2)
sp3 = spawn(task3)
# sp1.start()
# sp2.start()
# sp3.start()
# sp1.join()
# sp2.join()
# sp3.join()
joinall([sp1, sp2, sp3]) end_time = time.time() print(f'消耗时间: {end_time - start_time}')
TCP服务端实现协程
# server
from gevent import monkey; monkey.patch_all()
from gevent import spawn
import socket server = socket.socket() server.bind(
('127.0.0.1', 9999)
) server.listen(5) # 与客户端通信
def working(conn):
while True:
try:
data = conn.recv(1024)
if len(data) == 0:
break print(data.decode('utf-8')) conn.send(data.upper()) except Exception as e:
print(e)
break conn.close() # 与客户端连接
def run(server):
while True:
conn, addr = server.accept()
print(addr)
spawn(working, conn) if __name__ == '__main__':
print('服务端已启动...')
g = spawn(run, server)
g.join()
#client
from threading import Thread, current_thread
import socket def send_get_msg():
client = socket.socket() client.connect(
('127.0.0.1', 9999)
)
while True:
client.send(f'{current_thread().name}'.encode('utf-8'))
data = client.recv(1024)
print(data.decode('utf-8')) # 模拟100个用户访问服务端
if __name__ == '__main__':
list1 = []
for line in range(100):
t = Thread(target=send_get_msg)
t.start()
list1.append(t) for t in list1:
t.join()