不能无限的开进程，不能无限的开线程
最常用的就是开进程池，开线程池。其中回调函数非常重要
回调函数其实可以作为一种编程思想，谁好了谁就去调
只要你用并发，就会有锁的问题，但是你不能一直去自己加锁吧
那么我们就用QUEUE，这样还解决了自动加锁的问题
由Queue延伸出的一个点也非常重要的概念。以后写程序也会用到
这个思想。就是生产者与消费者问题
一、Python标准模块--concurrent.futures（并发未来）
concurent.future模块需要了解的
1.concurent.future模块是用来创建并行的任务，提供了更高级别的接口，
为了异步执行调用
2.concurent.future这个模块用起来非常方便，它的接口也封装的非常简单
3.concurent.future模块既可以实现进程池，也可以实现线程池
4.模块导入进程池和线程池
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
5.p = ProcessPoolExecutor(max_works)对于进程池如果不写max_works：默认的是cpu的数目,默认是4个
p = ThreadPoolExecutor(max_works)对于线程池如果不写max_works：默认的是cpu的数目*5
6.如果是进程池，得到的结果如果是一个对象。我们得用一个.get()方法得到结果
但是现在用了concurent.future模块，我们可以用obj.result方法
p.submit(task,i) #相当于apply_async异步方法
p.shutdown() #默认有个参数wite=True (相当于close和join)
二、线程池
进程池：就是在一个进程内控制一定个数的线程
基于concurent.future模块的进程池和线程池 （他们的同步执行和异步执行是一样的）
线程池内生成一定数量的线程,当遇到I/O时切换.并不是说有多少个任务,就就多少个线程.
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
import os
import time
import random
# I/O密集型的用线程池,用进程池的话开销大,效率低
#----------------同步执行-----------------
def task(n):
print("%s is runing"%os.getpid())
time.sleep(random.randint(1,3))
return n**2 if __name__ == '__main__':
temp_li=[]
start = time.time()
p=ProcessPoolExecutor(max_workers=4)
for i in range(10):
obj=p.submit(task,i).result() #等待结果,相当于apple同步方法.永远都会只是4个进程,就算有100个任务,也是4个进程轮流切换
temp_li.append(obj)
p.shutdown() #相当于close和join
print(temp_li)
print("耗时:%s"%(time.time()-start)) #---------------异步执行------------------
def task(n):
print("%s is running"%os.getpid())
time.sleep(random.randint(1,3))
return n**2 if __name__ == '__main__':
temp_li=[]
start=time.time()
p=ProcessPoolExecutor(max_workers=4) #如果不填写max_workers,默认是cpu核数
for i in range(10):
obj = p.submit(task,i) #不等待结果,提交完就走
print(obj) #打印进程状态的话,会看到有些是running(运行态),有些是pending(就绪).
temp_li.append(obj)
p.shutdown()
print([obj.result() for obj in temp_li])
print("耗时:%s"%(time.time() - start))
基于concurrent.future进程池
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
import os
import time
import random def task(n):
print("%s is running"%os.getpid())
time.sleep(random.randint(1,3))
return n**2 if __name__ == '__main__':
temp_li=[]
start=time.time()
t=ThreadPoolExecutor() #如果不填写max_workers,默认是cpu核数*5
for i in range(10):
obj =t.submit(task,i) #不等待结果,提交完就走
print(obj) #打印进程状态的话,会看到有些是running(运行态),有些是pending(就绪).
temp_li.append(obj)
t.shutdown()
print([obj.result() for obj in temp_li])
print("耗时:%s"%(time.time() - start)) #3.003171682357788
基于concurrent.future线程池
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import requests
import time
import os def get_page(url):
print("%s is getting %s"%(os.getpid(),url))
response = requests.get(url)
if response.status_code==200:#200表示下载成功的状态码
return {"url":url,"text":response.text} def pares_page(res):
res = res.result()
print("%s is getting %s"%(os.getpid(),res["url"]))
with open("db1.txt","a",encoding="utf-8") as f:
pares_res = "url:%s size:%s \n" %(res["url"],len(res["url"]))
f.write(pares_res) if __name__ == '__main__':
p = ProcessPoolExecutor()
# p = ThreadPoolExecutor()
li =[
"http://www.baidu.com",
"http://www.google.com",
"http://www.youporn.com"
] for url in li:
res = p.submit(get_page,url).add_done_callback(pares_page)#回调函数.
p.shutdown()
print("main",os.getpid())
线程池应用
map函数的应用
# map函数举例
obj= map(lambda x:x**2 ,range(10))
print(list(obj)) #运行结果[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
#! -*- coding:utf-8 -*-
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
import os
import time
import random
def task(n):
print("%s is running"%os.getpid())
time.sleep(random.randint(1,3))
return n**2 if __name__ == '__main__':
temp_li=[]
start=time.time()
t=ThreadPoolExecutor() #如果不填写max_workers,默认是cpu核数*5
obj = t.map(task,range(10))
t.shutdown() print(list(obj))
print("耗时:%s"%(time.time() - start))
map函数
三,协程
协程是一种用户态的轻量级线程，协程的调度完全由用户控制。协程拥有自己的寄存器上下文和栈。协程调度切换时，将寄存器上下文和栈保存到其他地方，在切回来的时候，恢复先前保存的寄存器上下文和栈，直接操作栈则基本没有内核切换的开销，可以不加锁的访问全局变量，所以上下文的切换非常快。
优点:
1. 协程的切换开销更小，属于程序级别的切换，操作系统完全感知不到，因而更加轻量级
2. 单线程内就可以实现并发的效果，最大限度地利用cpu
缺点:
1. 协程的本质是单线程下，无法利用多核，可以是一个程序开启多个进程，每个进程内开启多个线程，每个线程内开启协程
2. 协程指的是单个线程，因而一旦协程出现阻塞，将会阻塞整个线程
总结协程特点：
- 必须在只有一个单线程里实现并发
- 修改共享数据不需加锁
- 用户程序里自己保存多个控制流的上下文栈
- 附加：一个协程遇到IO操作自动切换到其它协程（如何实现检测IO，yield、greenlet都无法实现，就用到了gevent模块（select机制））
四,greenlet模块
如果我们现在有一个单线程,里面有10个任务,那么如果我们使用yield生成器来实现的话,太麻烦了(初始化生成器,调用send).这就用到了greenlet模块了.
Greenlet模块和yield没有什么区别，就只是单纯的切，跟效率无关。只不过比yield更好一点，切的时候方便一点。但是仍然没有解决效率.Greenlet可以让你在多个任务之间来回的切.
安装模块:
pip3 install greenlet
#! -*- coding:utf-8 -*-
from greenlet import greenlet def eat(name):
print("%s is eating 1"%name)
g2.switch("jack")
print("%s is eating 2"%name)
g2.switch() def running(name):
print("%s is running"%name)
g1.switch()
print("%s is not running"%name) if __name__ == '__main__':
g1 = greenlet(eat)
g2= greenlet(running) g1.switch("alex")
greenlet例子
单纯的切换（在没有io的情况下或者没有重复开辟内存空间的操作），反而会降低程序的执行速度.
import time
def f1():
res=1
for i in range(100000000):
res+=i def f2():
res=1
for i in range(100000000):
res*=i start=time.time()
f1()
f2()
stop=time.time()#10.354592561721802 from greenlet import greenlet
import time
def f1():
res=1
for i in range(100000000):
res+=i
g2.switch() def f2():
res=1
for i in range(100000000):
res*=i
g1.switch() start=time.time()
g1=greenlet(f1)
g2=greenlet(f2)
g1.switch()
stop=time.time()
print('run time is %s' %(stop-start)) # 51.55694890022278
单纯的切换,反而降低效率
greenlet只是提供了一种比generator更加便捷的切换方式，当切到一个任务执行时如果遇到io，那就原地阻塞，仍然是没有解决遇到IO自动切换来提升效率的问题。单线程里的这10个任务的代码通常会既有计算操作又有阻塞操作，我们完全可以在执行任务1时遇到阻塞，就利用阻塞的时间去执行任务2。。。。如此，才能提高效率，这就用到了Gevent模块。
五,gevnet模块
安装:
pip3 install gevent
Gevent是一个第三方库，可以轻松通过gevent实现并发同步或异步编程，在gevent中用到的主要模式是Greenlet,它是以C扩展模块形式接入Python的轻量级协程。Greenlet全部运行在主程序操作系统进程的内部，但它们被协作式地调度。
用法:
import gevent def test(name):
print("%s is eating"%name)
return 1111 def test2(name):
print("%s is going"%name)
return 3333 # g1 = gevent.spawn(函数名,位置参数(*args),关键字参数(**kwargs))
g1 = gevent.spawn(test,"jack")
g2 = gevent.spawn(test2,"alex")
gevent.joinall([g1,g2]) #等待g1,g2结束,也可以写成单个g1.join(),g2.join()
# g1.join()
print(g1.value) #拿到返回值
print(g2.value)
gevnet的一些方法:
# from gevent import monkey;monkey.patch_all()
import gevent
import time def eat(name):
print("%s is eat"%name)
# time.sleep(1.5) #模拟IO阻塞
"""
如果使用time.sleep()表示时间等待的话,需要在代码顶部加入一行 from gevent import monkey;monkey.patch_all()
如果使用gevent.sleep()则无需加入代码 """
gevent.sleep(1.5)
print("%s is eat 1"%name)
return "eat" def play(name):
print('%s play 1'%name)
# time.sleep(3)
gevent.sleep(3)
print('%s play 2'%name)
return 'paly' # 当有返回值的时候，gevent模块也提供了返回结果的操作 start_time = time.time()
g1 = gevent.spawn(eat,"jack")
g2 = gevent.spawn(play,"Lucy")
gevent.joinall([g1,g2])
print("main",time.time()-start_time)
print(g1.value)
print(g2.value)
注意time.sleep()和gevent.sleep()
注意:
gevent.sleep(1.5)模拟的是gevent可以识别的io阻塞,
而time.sleep(1.5)或其他的阻塞,gevent是不能直接识别的需要用下面一行代码,打补丁,就可以识别了
from gevent import monkey;monkey.patch_all()必须放到被打补丁者的前面，如time，socket模块之前
或者我们干脆记忆成：要用gevent，需要将from gevent import monkey;monkey.patch_all()放到文件的开头
from gevent import monkey;monkey.patch_all()
import gevent
import time def task(pid):
time.sleep(0.5)
print("Task %s is done" % pid) def synchronous(): #同步
for i in range(10):
task(i) def asynchronoues(): #异步
g_l = [gevent.spawn(task,i) for i in range(10)]
print(g_l)
gevent.joinall(g_l) if __name__ == '__main__':
print("sync")
synchronous()
print("async")
asynchronoues() #上面程序的重要部分是将task函数封装到Greenlet内部线程的gevent.spawn。 初始化的greenlet列表存放在数组threads中，此数组被传给gevent.joinall 函数，后者阻塞当前流程，并执行所有给定的greenlet。执行流程只会在 所有greenlet执行完后才会继续向下走。
同步异步