投资网站建设公司多少钱,临平做网站电话,全程电子化服务平台,h5制作平台教程一. 事件循环1.注#xff1a;实现搭配#xff1a;事件循环回调(驱动生成器【协程】)epoll(IO多路复用)#xff0c;asyncio是Python用于解决异步编程的一整套解决方案#xff1b;基于asynico#xff1a;tornado#xff0c;gevent#xff0c;twisted(Scrapy#xff0c;dj…一. 事件循环1.注实现搭配事件循环回调(驱动生成器【协程】)epoll(IO多路复用)asyncio是Python用于解决异步编程的一整套解决方案基于asynicotornadogeventtwisted(Scrapydjango channels)tornado(实现了web服务器可以直接部署真正部署还是要加nginx)djangoflask(uwsgigunicornnginx部署)1 importasyncio2 importtime3 async defget_html(url):4 print(start get url)5 #不能直接使用time.sleep,这是阻塞的函数如果使用time在并发的情况有多少个就有多少个2秒6 await asyncio.sleep(2)7 print(end get url)8 if __name____main__:9 start_timetime.time()10 loopasyncio.get_event_loop()11 task[get_html(www.baidu.com) for i in range(10)]12 loop.run_until_complete(asyncio.wait(task))13 print(time.time()-start_time)View Code2.如何获取协程的返回值(和线程池类似):1 importasyncio2 importtime3 from functools importpartial4 async defget_html(url):5 print(start get url)6 await asyncio.sleep(2)7 print(end get url)8 return HAHA9 #需要接收task如果要接收其他的参数就需要用到partial(偏函数),参数需要放到前面10 defcallback(url,future):11 print(urlsuccess)12 print(send email)13 if __name____main__:14 loopasyncio.get_event_loop()15 taskloop.create_task(get_html(www.baidu.com))16 #原理还是获取event_loop然后调用create_task方法一个线程只有一个loop17 #get_futureasyncio.ensure_future(get_html(www.baidu.com))也可以18 #loop.run_until_complete(get_future)19 #run_until_complete可以接收future类型task类型(是future类型的一个子类)也可以接收可迭代类型20 task.add_done_callback(partial(callback,www.baidu.com))21 loop.run_until_complete(task)22 print(task.result())View Code3.wait和gather的区别:3.1wait简单使用1 importasyncio2 importtime3 from functools importpartial4 async defget_html(url):5 print(start get url)6 await asyncio.sleep(2)7 print(end get url)89 if __name____main__:10 loopasyncio.get_event_loop()11 tasks[get_html(www.baidu.com) for i in range(10)]12 #wait和线程的wait相似13 loop.run_until_complete(asyncio.wait(tasks))View Code协程的wait和线程的wait相似也有timeoutreturn_when(什么时候返回)等参数3.2gather简单使用1 importasyncio2 importtime3 from functools importpartial4 async defget_html(url):5 print(start get url)6 await asyncio.sleep(2)7 print(end get url)89 if __name____main__:10 loopasyncio.get_event_loop()11 tasks[get_html(www.baidu.com) for i in range(10)]12 #gather注意加*这样就会变成参数13 loop.run_until_complete(asyncio.gather(*tasks))View Code3.3gather和wait的区别(定制性不强时可以优先考虑gather)gather更加高层可以将tasks分组还可以成批的取消任务1 importasyncio2 importtime3 from functools importpartial4 async defget_html(url):5 print(start get url)6 await asyncio.sleep(2)7 print(end get url)89 if __name____main__:10 loopasyncio.get_event_loop()11 groups1[get_html(www.baidu.com) for i in range(10)]12 groups2[get_html(www.baidu.com) for i in range(10)]13 #gather注意加*这样就会变成参数14 loop.run_until_complete(asyncio.gather(*groups1,*groups2))15 #这种方式也可以16 #groups1 [get_html(www.baidu.com) for i in range(10)]17 #groups2 [get_html(www.baidu.com) for i in range(10)]18 #groups1asyncio.gather(*groups1)19 #groups2asyncio.gather(*groups2)20 #取消任务21 #groups2.cancel()22 #loop.run_until_complete(asyncio.gather(groups1,groups2))View Code二. 协程嵌套1.run_util_complete()源码和run_forever()区别并不大只是可以在运行完指定的协程后可以把loop停止掉而run_forever()不会停止2.loop会被放在future里面future又会放在loop中3.取消future(task)3.1子协程调用原理官网例子解释 await相当于yield fromloop运行协程print_sum(),print_sum又会去调用另一个协程computerun_util_complete会把协程print_sum注册到loop中。1.event_loop会为print_sum创建一个task通过驱动task执行print_sum(task首先会进入pending【等待】的状态)2.print_sum直接进入字协程的调度这个时候转向执行另一个协程(compute所以print_sum变为suspended【暂停】状态)3.compute这个协程首先打印然后去调用asyncio的sleep(此时compute进入suspende的状态【暂停】)直接把返回值返回给Task(没有经过print_sum相当于yield from直接在调用方和子生成器通信是由委托方print_sum建立的通道)4.Task会告诉Event_loop暂停Event_loop等待一秒后通过Task唤醒(越过print_sum和compute建立一个通道)5.compute继续执行变为状态done【执行完成】然后抛一个StopIteration的异常会被await语句捕捉到然后提取出123的值进入print_sum,print_sum也被激活(因为抛出了StopIteration的异常被print_sum捕捉)print_sum执行完也会被标记为done的状态同时抛出StopIteration会被Task接收三. call_soon、call_later、call_at、call_soon_threadsafe1.call_soon:可以直接接收函数而不用协程1 importasyncio2 #函数3 defcallback(sleep_time):4 print(sleep {} success.format(sleep_time))5 #通过该函数暂停6 defstoploop(loop):7 loop.stop()8 if __name____main__:9 loopasyncio.get_event_loop()10 #可以直接传递函数而不用协程,call_soon其实就是调用的call_later时间为0秒11 loop.call_soon(callback,2)12 loop.call_soon(stoploop,loop)13 #不能用run_util_complete(因为不是协程)run_forever找到call_soon一直运行14 loop.run_forever()View Code2.call_later:可以指定多长时间后启动(实际调用call_at,时间不是传统的时间而是loop内部的时间)1 importasyncio2 #函数3 defcallback(sleep_time):4 print(sleep {} success.format(sleep_time))5 #通过该函数暂停6 defstoploop(loop):7 loop.stop()8 if __name____main__:9 loopasyncio.get_event_loop()10 loop.call_later(3,callback,1)11 loop.call_later(1, callback, 2)12 loop.call_later(1, callback, 2)13 loop.call_later(1, callback, 2)14 loop.call_soon(callback,4)15 #loop.call_soon(stoploop,loop)16 #不能用run_util_complete(因为不是协程)run_forever找到call_soon一直运行17 loop.run_forever()View Code3.call_at:指定某个时间执行1 importasyncio2 #函数3 defcallback(sleep_time):4 print(sleep {} success.format(sleep_time))5 #通过该函数暂停6 defstoploop(loop):7 loop.stop()8 if __name____main__:9 loopasyncio.get_event_loop()10 nowloop.time()11 print(now)12 loop.call_at(now3,callback,1)13 loop.call_at(now1, callback, 0.5)14 loop.call_at(now1, callback, 2)15 loop.call_at(now1, callback, 2)16 #loop.call_soon(stoploop,loop)17 #不能用run_util_complete(因为不是协程)run_forever找到call_soon一直运行18 loop.run_forever()View Code4.call_soon_threadsafe:线程安全的方法不仅能解决协程也能解决线程进程和call_soon几乎一致多了self._write_to_self()和call_soon用法一致四. ThreadPoolExecutorasyncio(线程池和协程结合)1.使用run_in_executor就是把阻塞的代码放进线程池运行性能并不是特别高和多线程差不多1 #使用多线程在协程中集成阻塞io2 importasyncio3 importsocket4 from urllib.parse importurlparse5 from concurrent.futures importThreadPoolExecutor6 importtime7 defget_url(url):8 #通过socket请求html9 urlurlparse(url)10 hosturl.netloc11 pathurl.path12 if path:13 path/14 #建立socket连接15 clientsocket.socket(socket.AF_INET,socket.SOCK_STREAM)16 client.connect((host,80))17 #向服务器发送数据18 client.send(GET {} HTTP/1.1\r\nHost:{}\r\nConnection:close\r\n\r\n.format(path, host).encode(utf8))19 #将数据读取完20 datab21 whileTrue:22 dclient.recv(1024)23 ifd:24 datad25 else:26 break27 #会将header信息作为返回字符串28 datadata.decode(utf8)29 print(data.split(\r\n\r\n)[1])30 client.close()3132 if __name____main__:33 start_timetime.time()34 loopasyncio.get_event_loop()35 excutorThreadPoolExecutor()36 tasks[]37 for i in range(20):38 taskloop.run_in_executor(excutor,get_url,http://www.baidu.com)39 tasks.append(task)40 loop.run_until_complete(asyncio.wait(tasks))41 print(time.time()-start_time)View Code五. asyncio模拟http请求注asyncio目前没有提供http协议的接口1 #asyncio目前没有提供http协议的接口2 importasyncio3 from urllib.parse importurlparse4 importtime567 async defget_url(url):8 #通过socket请求html9 url urlparse(url)10 host url.netloc11 path url.path12 if path :13 path /14 #建立socket连接(比较耗时)非阻塞需要注册都在open_connection中实现了15 reader, writer await asyncio.open_connection(host, 80)16 #向服务器发送数据,unregister和register都实现了17 writer.write(GET {} HTTP/1.1\r\nHost:{}\r\nConnection:close\r\n\r\n.format(path, host).encode(utf8))18 #读取数据19 all_lines []20 #源码实现较复杂有__anext__的魔法函数(协程)21 async for line inreader:22 data line.decode(utf8)23 all_lines.append(data)24 html \n.join(all_lines)25 returnhtml262728 async defmain():29 tasks []30 for i in range(20):31 url http://www.baidu.com/32 tasks.append(asyncio.ensure_future(get_url(url)))33 for task inasyncio.as_completed(tasks):34 result await task35 print(result)363738 if __name__ __main__:39 start_time time.time()40 loop asyncio.get_event_loop()41 #tasks[get_url(http://www.baidu.com) for i in range(10)]42 #在外部获取结果,保存为future对象43 #tasks [asyncio.ensure_future(get_url(http://www.baidu.com)) for i in range(10)]44 #loop.run_until_complete(asyncio.wait(tasks))45 #for task in tasks:46 #print(task.result())47 #执行完一个打印一个48 loop.run_until_complete(main())49 print(time.time() - start_time)View Code六. future和task1.future协程中的future和线程池中的future相似future中的方法都和线程池中的相似set_result方法不像线程池中运行完直接运行代码(这是单线程的会调用call_soon方法)2.task是future的子类是future和协程之间的桥梁会首先启动_step方法该方法会首先启动协程把返回值(StopIteration的值)做处理用于解决协程和线程不一致的地方七. asyncio同步和通信1.单线程协程不需要锁1 importasyncio2 total03 async defadd():4 globaltotal5 for i in range(1000000):6 total1789 async defdecs():10 globaltotal11 for i in range(1000000):12 total-113 if __name____main__:14 loopasyncio.get_event_loop()15 tasks[add(),decs()]16 loop.run_until_complete(asyncio.wait(tasks))17 print(total)View Code2.某种情况需要锁asyncio中的锁(同步机制)1 importasyncio,aiohttp2 #这是并没有调用系统的锁只是简单的自己实现(注意是非阻塞的),Queue也是非阻塞的,都用了yield from不用用到condition【单线程】】3 #Queue还可以限流如果只需要通信还可以直接使用全局变量否则可以4 from asyncio importLock,Queue5 catche{}6 lockLock()7 async defget_stuff():8 #实现了__enter__和__exit__两个魔法函数可以用with9 #with await lock:10 #更明确的语法__aenter__和__await__11 async with lock:12 #注意加await,是一个协程13 #await lock.acquire()14 for url incatche:15 returncatche[url]16 #异步的接收17 stauffaiohttp.request(Get,url)18 catche[url]stauff19 returnstauff20 #release是一个简单的函数21 #lock.release()2223 async defparse_stuff():24 stuffawait get_stuff()2526 async defuse_stuff():27 stuffawait get_stuff()28 #如果没有同步机制就会发起两次请求(这里就可以加一个同步机制)29 tasks[parse_stuff(),use_stuff()]30 loopasyncio.get_event_loop()31 loop.run_until_complete(asyncio.wait(tasks))View Code八. aiohttp实现高并发爬虫1 #asyncio去重url入库(异步的驱动aiomysql)2 importaiohttp3 importasyncio4 importre5 importaiomysql6 from pyquery importpyquery78 start_url http://www.jobbole.com/9 waiting_urls []10 seen_urls []11 stopping False12 #限制并发数13 semasyncio.Semaphore(3)141516 async deffetch(url, session):17 async with sem:18 await asyncio.sleep(1)19 try:20 async with session.get(url) as resp:21 print(url_status:{}.format(resp.status))22 if resp.status in [200, 201]:23 data await resp.text()24 returndata25 exceptException as e:26 print(e)272829 defextract_urls(html):30 31 解析无io操作32 33 urls []34 pq pyquery(html)35 for link in pq.items(a):36 url link.attr(href)37 if url and url.startwith(http) and url not inurls:38 urls.append(url)39 waiting_urls.append(url)40 returnurls414243 async definit_urls(url, session):44 html await fetch(url, session)45 seen_urls.add(url)46 extract_urls(html)474849 async defhandle_article(url, session, pool):50 51 处理文章52 53 html await fetch(url, session)54 seen_urls.append(url)55 extract_urls(html)56 pq pyquery(html)57 title pq(title).text()58 async with pool.acquire() as conn:59 async with conn.cursor() as cur:60 insert_sql insert into Test(title) values({}).format(title)61 await cur.execute(insert_sql)626364 async defconsumer(pool):65 with aiohttp.CLientSession() as session:66 while notstopping:67 if len(waiting_urls) 0:68 await asyncio.sleep(0.5)69 continue70 url waiting_urls.pop()71 print(start url: url)72 if re.match(http://.*?jobble.com/\d/, url):73 if url not inseen_urls:74 asyncio.ensure_future(handle_article(url, session, pool))75 await asyncio.sleep(30)76 else:77 if url not inseen_urls:78 asyncio.ensure_future(init_urls(url, session))798081 async defmain():82 #等待mysql连接好83 pool aiomysql.connect(hostlocalhost, port3306, userroot,84 password112358, dbmy_aio, looploop, charsetutf8, autocommitTrue)85 async with aiohttp.CLientSession() as session:86 html await fetch(start_url, session)87 seen_urls.add(start_url)88 extract_urls(html)89 asyncio.ensure_future(consumer(pool))9091 if __name__ __main__:92 loop asyncio.get_event_loop()93 asyncio.ensure_future(loop)94 loop.run_forever(main(loop))View Code
