舟山 做企业网站,做网站建设找哪家好,wordpress 编辑器 空白,湖北广盛建设集团网站需要注意一下不能无限的开进程#xff0c;不能无限的开线程最常用的就是开进程池#xff0c;开线程池。其中回调函数非常重要回调函数其实可以作为一种编程思想#xff0c;谁好了谁就去掉只要你用并发#xff0c;就会有锁的问题#xff0c;但是你不能一直去自己加锁吧那么…需要注意一下不能无限的开进程不能无限的开线程最常用的就是开进程池开线程池。其中回调函数非常重要回调函数其实可以作为一种编程思想谁好了谁就去掉只要你用并发就会有锁的问题但是你不能一直去自己加锁吧那么我们就用QUEUE这样还解决了自动加锁的问题由Queue延伸出的一个点也非常重要的概念。以后写程序也会用到这个思想。就是生产者与消费者问题一、Python标准模块--concurrent.futures(并发未来)concurent.future模块需要了解的1.concurent.future模块是用来创建并行的任务提供了更高级别的接口为了异步执行调用2.concurent.future这个模块用起来非常方便它的接口也封装的非常简单3.concurent.future模块既可以实现进程池也可以实现线程池4.模块导入进程池和线程池from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor还可以导入一个Executor,但是你别这样导这个类是一个抽象类抽象类的目的是规范他的子类必须有某种方法(并且抽象类的方法必须实现)但是抽象类不能被实例化5.p ProcessPoolExecutor(max_works)对于进程池如果不写max_works默认的是cpu的数目,默认是4个p ThreadPoolExecutor(max_works)对于线程池如果不写max_works默认的是cpu的数目*56.如果是进程池得到的结果如果是一个对象。我们得用一个.get()方法得到结果但是现在用了concurent.future模块我们可以用obj.result方法p.submit(task,i) #相当于apply_async异步方法p.shutdown() #默认有个参数witeTrue (相当于close和join)那么什么是线程池呢我们来了解一下二、线程池进程池就是在一个进程内控制一定个数的线程基于concurent.future模块的进程池和线程池 (他们的同步执行和异步执行是一样的)1 #1.同步执行--------------2 from concurrent.futures importProcessPoolExecutor,ThreadPoolExecutor3 importos,time,random4 deftask(n):5 print([%s] is running%os.getpid())6 time.sleep(random.randint(1,3)) #I/O密集型的一般用线程用了进程耗时长7 return n**28 if __name__ __main__:9 start time.time()10 p ProcessPoolExecutor()11 for i in range(10): #现在是开了10个任务 那么如果是上百个任务呢就不能无线的开进程那么就得考虑控制12 #线程数了那么就得考虑到池了13 obj p.submit(task,i).result() #相当于apply同步方法14 p.shutdown() #相当于close和join方法15 print(*30)16 print(time.time() - start) #17.36499309539795171819 #2.异步执行-----------20 #from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor21 #import os,time,random22 #def task(n):23 #print([%s] is running%os.getpid())24 #time.sleep(random.randint(1,3)) #I/O密集型的一般用线程用了进程耗时长25 #return n**226 #if __name__ __main__:27 #start time.time()28 #p ProcessPoolExecutor()29 #l []30 #for i in range(10): #现在是开了10个任务 那么如果是上百个任务呢就不能无线的开进程那么就得考虑控制31 ## 线程数了那么就得考虑到池了32 #obj p.submit(task,i) #相当于apply_async()异步方法33 #l.append(obj)34 #p.shutdown() #相当于close和join方法35 #print(*30)36 #print([obj.result() for obj in l])37 #print(time.time() - start) #5.362306594848633基于concurrent.futures模块的进程池1 from concurrent.futures importProcessPoolExecutor,ThreadPoolExecutor2 from threading importcurrentThread3 importos,time,random4 deftask(n):5 print(%s:%s is running%(currentThread().getName(),os.getpid())) #看到的pid都是一样的因为线程是共享了一个进程6 time.sleep(random.randint(1,3)) #I/O密集型的一般用线程用了进程耗时长7 return n**28 if __name__ __main__:9 start time.time()10 p ThreadPoolExecutor() #线程池 #如果不给定值默认cup*511 l []12 for i in range(10): #10个任务 # 线程池效率高了13 obj p.submit(task,i) #相当于apply_async异步方法14l.append(obj)15 p.shutdown() #默认有个参数witeTrue (相当于close和join)16 print(*30)17 print([obj.result() for obj inl])18 print(time.time() - start) #3.001171827316284基于concurrent.futures模块的线程池应用线程池(下载网页并解析)from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutorimport requestsimport time,osdef get_page(url):print( is getting [%s]%(os.getpid(),url))response requests.get(url)if response.status_code200: #200代表状态下载成功了return {url:url,text:response.text}def parse_page(res):res res.result()print( is getting [%s]%(os.getpid(),res[url]))with open(db.txt,a) as f:parse_res url:%s size:%s\n%(res[url],len(res[text]))f.write(parse_res)if __name__ __main__:# p ThreadPoolExecutor()p ProcessPoolExecutor()l [http://www.baidu.com,http://www.baidu.com,http://www.baidu.com,http://www.baidu.com,]for url in l:res p.submit(get_page,url).add_done_callback(parse_page) #这里的回调函数拿到的是一个对象。得# 先把返回的res得到一个结果。即在前面加上一个res.result() #谁好了谁去掉回调函数# 回调函数也是一种编程思想。不仅开线程池用开线程池也用p.shutdown() #相当于进程池里的close和joinprint(主,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]可以和上面的开进程池/线程池的对比着看就能发现map函数的强大了map函数的应用三、协程介绍协程单线程下实现并发(提高效率)说到协成我们先说一下协程联想到的知识点yield复习1 3.yield功能2(可以吧函数暂停住保存原来的状态)--------------2 deff1():3 print(first)4 yield 15 print(second)6 yield 27 print(third)8 yield 39 #print(f1()) #加了yield返回的是一个生成器10 g f1()11 print(next(g)) #当遇见了yield的时候就返回一个值,而且保存原来的状态12 print(next(g)) #当遇见了yield的时候就返回一个值13 print(next(g)) #当遇见了yield的时候就返回一个值yield功能示例11 #3.yield表达式(对于表达式的yield)--------------------2 importtime3 defwrapper(func):4 def inner(*args,**kwargs):5 ret func(*args,**kwargs)6next(ret)7 returnret8 returninner9wrapper10 defconsumer():11 whileTrue:12 x yield13 print(x)1415 defproducter(target):16 生产者造值17 #next(g) #相当于g.send(None)18 for i in range(10):19 time.sleep(0.5)20 target.send(i)#要用send就得用两个yield21 producter(consumer())yield功能示例2引子本节主题是实现单线程下的并发即只在一个主线程并且很明显的是可利用的cpu只有一个情况下实现并发为此我们需要先回顾下并发的本质切换保存状态cpu正在运行一个任务会在两种情况下切走去执行其他的任务(切换由操作系统强制控制)一种情况是该任务发生了阻塞另外一种情况是该任务计算的时间过长其中第二种情况并不能提升效率只是为了让cpu能够雨露均沾实现看起来大家都被执行的效果如果多个程序都是纯计算任务这种切换反而会降低效率。为此我们可以基于yield来验证。yield本身就是一种在单线程下可以保存任务运行状态的方法我们来简单复习一下单纯的切反而会影响效率1 #串行执行2 importtime3 defconsumer(res):4 任务1:接收数据,处理数据5 pass67 defproducer():8 任务2:生产数据9 res[]10 for i in range(10000000):11res.append(i)12 returnres1314 starttime.time()15 #串行执行16 resproducer()17consumer(res)18 stoptime.time()19 print(stop-start) #1.5536692142486572串行执行1 importtime2 defwrapper(func):3 def inner(*args,**kwargs):4 ret func(*args,**kwargs)5next(ret)6 returnret7 returninner8wrapper9 defconsumer():10 whileTrue:11 x yield12 print(x)1314 defproducter(target):15 生产者造值16 #next(g) #相当于g.send(None)17 for i in range(10):18 time.sleep(0.5)19 target.send(i)#要用send就得用两个yield20 producter(consumer())基于yield并发执行对于单线程下我们不可避免程序中出现io操作但如果我们能在自己的程序中(即用户程序级别而非操作系统级别)控制单线程下多个任务能遇到io就切换这样就保证了该线程能够最大限度地处于就绪态即随时都可以被cpu执行的状态相当于我们在用户程序级别将自己的io操作最大限度地隐藏起来对于操作系统来说这哥们(该线程)好像是一直处于计算过程的io比较少。协程的本质就是在单线程下由用户自己控制一个任务遇到io阻塞了就切换另外一个任务去执行以此来提升效率。因此我们需要找寻一种可以同时满足以下条件的解决方案1. 可以控制多个任务之间的切换切换之前将任务的状态保存下来(重新运行时可以基于暂停的位置继续)2. 作为1的补充可以检测io操作在遇到io操作的情况下才发生切换四、GreenletGreenlet模块和yield没有什么区别就只是单纯的切跟效率无关。只不过比yield更好一点切的时候方便一点。但是仍然没有解决效率Greenlet可以让你在多个任务之间来回的切举例1 from greenlet importgreenlet2 importtime3 defeat(name):4 print(%s eat 1 %name)5 time.sleep(10) #当遇到IO的时候它也没有切这就得用gevent了6 g2.switch(egon)7 print(%s eat 2 %name)8g2.switch()9 defplay(name):10 print(%s play 1 %name)11g1.switch()12 print(%s play 2 %name)1314 g1greenlet(eat)15 g2greenlet(play)1617 g1.switch(egon)#可以在第一次switch时传入参数以后都不需要greenlet所以上面的方法都不可行那么这就用到了Gevert ,也就是协程。就解决了单线程实现并发的问题还提升了效率五、Gevent介绍Gevent 是一个第三方库可以轻松通过gevent实现并发同步或异步编程在gevent中用到的主要模式是Greenlet,它是以C扩展模块形式接入Python的轻量级协程。 Greenlet全部运行在主程序操作系统进程的内部但它们被协作式地调度。#用法g1gevent.spawn(func,1,,2,3,x4,y5)创建一个协程对象g1spawn括号内第一个参数是函数名如eat后面可以有多个参数可以是位置实参或关键字实参都是传给函数eat的g2gevent.spawn(func2)g1.join() #等待g1结束g2.join() #等待g2结束#或者上述两步合作一步gevent.joinall([g1,g2])g1.value#拿到func1的返回值举例;1 from gevent importmonkey;monkey.patch_all()2 importgevent3 importtime4 defeat(name):5 print(%s eat 1 %name)6 time.sleep(2) #我们用等待的时间模拟IO阻塞7 在gevent模块里面要用gevent.sleep(2)表示等待的时间8 然而我们经常用time.sleep()用习惯了那么有些人就想着9 可以用time.sleep()那么也不是不可以。要想用就得在10 最上面导入from gevent import monkey;monkey.patch_all()这句话11 如果不导入直接用time.sleep()就实现不了单线程并发的效果了1213 #gevent.sleep(2)14 print(%s eat 2 %name)15 return eat16 defplay(name):17 print(%s play 1 %name)18 time.sleep(3)19 #gevent.sleep(3)20 print(%s play 2 %name)21 return paly #当有返回值的时候gevent模块也提供了返回结果的操作2223 start time.time()24 g1 gevent.spawn(eat,egon) #执行任务25 g2 gevent.spawn(play,egon) #g1和g2的参数可以不一样26 #g1.join() #等待g127 #g2.join() #等待g228 #上面等待的两句也可以这样写29gevent.joinall([g1,g2])30 print(主,time.time()-start) #3.0011715888977053132 print(g1.value)33 print(g2.value)gevent的一些方法(重要)需要说明的是gevent.sleep(2)模拟的是gevent可以识别的io阻塞,而time.sleep(2)或其他的阻塞,gevent是不能直接识别的需要用下面一行代码,打补丁,就可以识别了from gevent import monkey;monkey.patch_all()必须放到被打补丁者的前面如timesocket模块之前或者我们干脆记忆成要用gevent需要将from gevent import monkey;monkey.patch_all()放到文件的开头六、Gevent之同步于异步1 from gevent importspawn,joinall,monkey;monkey.patch_all()23 importtime4 deftask(pid):5 6 Some non-deterministic task78 time.sleep(0.5)9 print(Task %s done %pid)101112 defsynchronous():13 for i in range(10):14task(i)1516 defasynchronous():17 g_l[spawn(task,i) for i in range(10)]18joinall(g_l)1920 if __name__ __main__:21 print(Synchronous:)22synchronous()2324 print(Asynchronous:)25asynchronous()26 #上面程序的重要部分是将task函数封装到Greenlet内部线程的gevent.spawn。 初始化的greenlet列表存放在数组threads中此数组被传给gevent.joinall 函数后者阻塞当前流程并执行所有给定的greenlet。执行流程只会在 所有greenlet执行完后才会继续向下走。View Code七、Gevent之应用举例一1 from gevent import monkey;monkey.patch_all() #打补丁2 importgevent3 importrequests4 importtime5 defget_page(url):6 print(get :%s%url)7 response requests.get(url)8 if response.status_code200: #下载成功的状态9 print(%d bytes received from:%s%(len(response.text),url))10 starttime.time()11gevent.joinall([12 gevent.spawn(get_page,http://www.baidu.com),13 gevent.spawn(get_page, https://www.yahoo.com/),14 gevent.spawn(get_page, https://github.com/),15])16 stop time.time()17 print(run time is %s %(stop-start))协程应用爬虫from gevent importjoinall,spawn,monkey;monkey.patch_all()importrequestsfrom threading importcurrent_threaddefparse_page(res):print(%s PARSE %s %(current_thread().getName(),len(res)))def get_page(url,callbackparse_page):print(%s GET %s %(current_thread().getName(),url))responserequests.get(url)if response.status_code 200:callback(response.text)if __name__ __main__:urls[https://www.baidu.com,https://www.taobao.com,https://www.openstack.org,]tasks[]for url inurls:tasks.append(spawn(get_page,url))joinall(tasks)协程应用爬虫(回调函数)八、Gevent之应用举例二也可以利用协程实现并发1 #!usr/bin/env python2 #-*- coding:utf-8 -*-3 from gevent importmonkey;monkey.patch_all()4 importgevent5 from socket import *6 print(start running...)7 deftalk(conn,addr):8 whileTrue:9 data conn.recv(1024)10 print(%s:%s %s%(addr[0],addr[1],data))11conn.send(data.upper())12conn.close()13 defserver(ip,duankou):14 server socket(AF_INET, SOCK_STREAM)15 server.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)16server.bind((ip,duankou))17 server.listen(5)18 whileTrue:19 conn,addr server.accept() #等待链接20 gevent.spawn(talk,conn,addr) #异步执行 (p Process(targettalk,args(coon,addr))21 #p.start())相当于开进程里的这两句22server.close()23 if __name__ __main__:24 server(127.0.0.1,8081)服务端利用协程1 #!usr/bin/env python2 #-*- coding:utf-8 -*-3 from multiprocessing importProcess4 from gevent importmonkey;monkey.patch_all()5 from socket import *6 defclient(ip,duankou):7 client socket(AF_INET, SOCK_STREAM)8client.connect((ip,duankou))9 whileTrue:10 client.send(hello.encode(utf-8))11 data client.recv(1024)12 print(data.decode(utf-8))13 if __name__ __main__:14 for i in range(100):15 p Process(targetclient,args((127.0.0.1,8081)))16 p.start()客户端开了100个进程
