asyncio並發編程

一. 事件循環

　　　1.注：

　　　　　 　實現搭配：事件循環+回調（驅動生成器【協程】）+epoll（IO多路復用），asyncio是Python用於解決異步編程的一整套解決方案；

　　　    基於asynico：tornado，gevent，twisted（Scrapy，django channels），tornado（實現了web服務器，可以直接部署，真正部署還是要加nginx），django，flask（uwsgi，gunicorn+nginx部署）

import asyncio
import time
async def get_html(url):
    print('start get url')
    #不能直接使用time.sleep,這是阻塞的函數，如果使用time在並發的情況有多少個就有多少個2秒
    await asyncio.sleep(2)
    print('end get url')
if __name__=='__main__':
    start_time=time.time()
    loop=asyncio.get_event_loop()
    task=[get_html('www.baidu.com') for i in range(10)]
    loop.run_until_complete(asyncio.wait(task))
    print(time.time()-start_time)

　　2.如何獲取協程的返回值（和線程池類似）:

import asyncio
import time
from functools import partial
async def get_html(url):
    print('start get url')
    await asyncio.sleep(2)
    print('end get url')
    return "HAHA"
#需要接收task，如果要接收其他的參數就需要用到partial(偏函數),參數需要放到前面
def callback(url,future):
    print(url+' success')
    print('send email')
if __name__=='__main__':
    loop=asyncio.get_event_loop()
    task=loop.create_task(get_html('www.baidu.com'))
    #原理還是獲取event_loop，然后調用create_task方法，一個線程只有一個loop
    # get_future=asyncio.ensure_future(get_html('www.baidu.com'))也可以
    #loop.run_until_complete(get_future)
    #run_until_complete可以接收future類型，task類型（是future類型的一個子類），也可以接收可迭代類型
    task.add_done_callback(partial(callback,'www.baidu.com'))
    loop.run_until_complete(task)
    print(task.result())

　　3.wait和gather的區別:

　　　　3.1wait簡單使用：

import asyncio
import time
from functools import partial
async def get_html(url):
    print('start get url')
    await asyncio.sleep(2)
    print('end get url')

if __name__=='__main__':
    loop=asyncio.get_event_loop()
    tasks=[get_html('www.baidu.com') for i in range(10)]
    #wait和線程的wait相似
    loop.run_until_complete(asyncio.wait(tasks))

協程的wait和線程的wait相似，也有timeout，return_when（什么時候返回）等參數

 

　　  3.2gather簡單使用：

import asyncio
import time
from functools import partial
async def get_html(url):
    print('start get url')
    await asyncio.sleep(2)
    print('end get url')

if __name__=='__main__':
    loop=asyncio.get_event_loop()
    tasks=[get_html('www.baidu.com') for i in range(10)]
    #gather注意加*，這樣就會變成參數
    loop.run_until_complete(asyncio.gather(*tasks))

　　3.3gather和wait的區別：（定制性不強時可以優先考慮gather）

　　　　gather更加高層，可以將tasks分組；還可以成批的取消任務

 

import asyncio
import time
from functools import partial
async def get_html(url):
    print('start get url')
    await asyncio.sleep(2)
    print('end get url')

if __name__=='__main__':
    loop=asyncio.get_event_loop()
    groups1=[get_html('www.baidu.com') for i in range(10)]
    groups2=[get_html('www.baidu.com') for i in range(10)]
    #gather注意加*，這樣就會變成參數
    loop.run_until_complete(asyncio.gather(*groups1,*groups2))
    #這種方式也可以
    # groups1 = [get_html('www.baidu.com') for i in range(10)]
    # groups2 = [get_html('www.baidu.com') for i in range(10)]
    # groups1=asyncio.gather(*groups1)
    # groups2=asyncio.gather(*groups2)
    #取消任務
    # groups2.cancel()
    # loop.run_until_complete(asyncio.gather(groups1,groups2))

 

二. 協程嵌套

　　1.run_util_complete()源碼：和run_forever()區別並不大，只是可以在運行完指定的協程后可以把loop停止掉，而run_forever()不會停止

 

　　2.loop會被放在future里面，future又會放在loop中

　　3.取消future(task)：

　　　　3.1子協程調用原理：

　　　　　　官網例子：

 

　　　　　　解釋： await相當於yield from，loop運行協程print_sum(),print_sum又會去調用另一個協程compute，run_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語句捕捉到，然后提取出1+2=3的值，進入print_sum,print_sum也被激活（因為拋出了StopIteration的異常被print_sum捕捉），print_sum執行完也會被標記為done的狀態，同時拋出StopIteration會被Task接收

三. call_soon、call_later、call_at、call_soon_threadsafe

　　1.call_soon:可以直接接收函數，而不用協程

import asyncio
#函數
def callback(sleep_time):
    print('sleep {} success'.format(sleep_time))
#通過該函數暫停
def stoploop(loop):
    loop.stop()
if __name__=='__main__':
    loop=asyncio.get_event_loop()
    #可以直接傳遞函數，而不用協程,call_soon其實就是調用的call_later，時間為0秒
    loop.call_soon(callback,2)
    loop.call_soon(stoploop,loop)
    #不能用run_util_complete（因為不是協程），run_forever找到call_soon一直運行
    loop.run_forever()

 

　　2.call_later:可以指定多長時間后啟動(實際調用call_at,時間不是傳統的時間，而是loop內部的時間)

import asyncio
#函數
def callback(sleep_time):
    print('sleep {} success'.format(sleep_time))
#通過該函數暫停
def stoploop(loop):
    loop.stop()
if __name__=='__main__':
    loop=asyncio.get_event_loop()
    loop.call_later(3,callback,1)
    loop.call_later(1, callback, 2)
    loop.call_later(1, callback, 2)
    loop.call_later(1, callback, 2)
    loop.call_soon(callback,4)
    # loop.call_soon(stoploop,loop)
    #不能用run_util_complete（因為不是協程），run_forever找到call_soon一直運行
    loop.run_forever()

 

　　3.call_at:指定某個時間執行

import asyncio
#函數
def callback(sleep_time):
    print('sleep {} success'.format(sleep_time))
#通過該函數暫停
def stoploop(loop):
    loop.stop()
if __name__=='__main__':
    loop=asyncio.get_event_loop()
    now=loop.time()
    print(now)
    loop.call_at(now+3,callback,1)
    loop.call_at(now+1, callback, 0.5)
    loop.call_at(now+1, callback, 2)
    loop.call_at(now+1, callback, 2)
    # loop.call_soon(stoploop,loop)
    #不能用run_util_complete（因為不是協程），run_forever找到call_soon一直運行
    loop.run_forever()

 

　　4.call_soon_threadsafe:

　　　　線程安全的方法，不僅能解決協程，也能解決線程，進程，和call_soon幾乎一致，多了self._write_to_self()，和call_soon用法一致

四. ThreadPoolExecutor+asyncio（線程池和協程結合）

　　1.使用run_in_executor：就是把阻塞的代碼放進線程池運行，性能並不是特別高，和多線程差不多

　　

#使用多線程，在協程中集成阻塞io
import asyncio
import socket
from urllib.parse import urlparse
from concurrent.futures import ThreadPoolExecutor
import time
def get_url(url):
    #通過socket請求html
    url=urlparse(url)
    host=url.netloc
    path=url.path
    if path=="":
        path="/"
    #建立socket連接
    client=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
    client.connect((host,80))
    #向服務器發送數據
    client.send("GET {} HTTP/1.1\r\nHost:{}\r\nConnection:close\r\n\r\n".format(path, host).encode("utf8"))
    #將數據讀取完
    data=b""
    while True:
        d=client.recv(1024)
        if d:
            data+=d
        else:
            break
    #會將header信息作為返回字符串
    data=data.decode('utf8')
    print(data.split('\r\n\r\n')[1])
    client.close()

if __name__=='__main__':
    start_time=time.time()
    loop=asyncio.get_event_loop()
    excutor=ThreadPoolExecutor()
    tasks=[]
    for i in range(20):
        task=loop.run_in_executor(excutor,get_url,'http://www.baidu.com')
        tasks.append(task)
    loop.run_until_complete(asyncio.wait(tasks))
    print(time.time()-start_time)

 

五. asyncio模擬http請求

　　注：asyncio目前沒有提供http協議的接口

# asyncio目前沒有提供http協議的接口
import asyncio
from urllib.parse import urlparse
import time


async def get_url(url):
    # 通過socket請求html
    url = urlparse(url)
    host = url.netloc
    path = url.path
    if path == "":
        path = "/"
    # 建立socket連接(比較耗時)，非阻塞需要注冊，都在open_connection中實現了
    reader, writer = await asyncio.open_connection(host, 80)
    # 向服務器發送數據,unregister和register都實現了
    writer.write("GET {} HTTP/1.1\r\nHost:{}\r\nConnection:close\r\n\r\n".format(path, host).encode("utf8"))
    # 讀取數據
    all_lines = []
    # 源碼實現較復雜，有__anext__的魔法函數（協程）
    async for line in reader:
        data = line.decode('utf8')
        all_lines.append(data)
    html = '\n'.join(all_lines)
    return html


async def main():
    tasks = []
    for i in range(20):
        url = "http://www.baidu.com/"
        tasks.append(asyncio.ensure_future(get_url(url)))
    for task in asyncio.as_completed(tasks):
        result = await task
        print(result)


if __name__ == '__main__':
    start_time = time.time()
    loop = asyncio.get_event_loop()
    # tasks=[get_url('http://www.baidu.com') for i in range(10)]
    # 在外部獲取結果,保存為future對象
    # tasks = [asyncio.ensure_future(get_url('http://www.baidu.com')) for i in range(10)]
    # loop.run_until_complete(asyncio.wait(tasks))
    # for task in tasks:
    #     print(task.result())
    # 執行完一個打印一個
    loop.run_until_complete(main())
    print(time.time() - start_time)

 

六. future和task

　　1.future：協程中的future和線程池中的future相似

future中的方法，都和線程池中的相似

set_result方法

 不像線程池中運行完直接運行代碼（這是單線程的，會調用call_soon方法）

　　

　　2.task：是future的子類，是future和協程之間的橋梁

 會首先啟動_step方法

該方法會首先啟動協程，把返回值（StopIteration的值）做處理，用於解決協程和線程不一致的地方

七. asyncio同步和通信

　　1.單線程協程不需要鎖：

import asyncio
total=0
async def add():
    global total
    for i in range(1000000):
        total+=1


async def decs():
    global total
    for i in range(1000000):
        total-=1
if __name__=='__main__':
    loop=asyncio.get_event_loop()
    tasks=[add(),decs()]
    loop.run_until_complete(asyncio.wait(tasks))
    print(total)

 

　　2.某種情況需要鎖：

　

asyncio中的鎖（同步機制）

 

 

import asyncio,aiohttp
#這是並沒有調用系統的鎖，只是簡單的自己實現（注意是非阻塞的）,Queue也是非阻塞的,都用了yield from，不用用到condition【單線程】】
#Queue還可以限流，如果只需要通信還可以直接使用全局變量否則可以
from asyncio import Lock,Queue
catche={}
lock=Lock()
async def get_stuff():
    #實現了__enter__和__exit__兩個魔法函數，可以用with
    # with await lock:
    #更明確的語法__aenter__和__await__
    async with lock:
        #注意加await,是一個協程
        #await lock.acquire()
        for url in catche:
            return catche[url]
        #異步的接收
        stauff=aiohttp.request('Get',url)
        catche[url]=stauff
        return stauff
        #release是一個簡單的函數
        #lock.release()

async def parse_stuff():
    stuff=await get_stuff()

async def use_stuff():
    stuff=await get_stuff()
#如果沒有同步機制，就會發起兩次請求（這里就可以加一個同步機制）
tasks=[parse_stuff(),use_stuff()]
loop=asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))

 

八. aiohttp實現高並發爬蟲

 

# asyncio去重url，入庫（異步的驅動aiomysql）
import aiohttp
import asyncio
import re
import aiomysql
from pyquery import pyquery

start_url = 'http://www.jobbole.com/'
waiting_urls = []
seen_urls = []
stopping = False
#限制並發數
sem=asyncio.Semaphore(3)


async def fetch(url, session):
    async with sem:
        await asyncio.sleep(1)
        try:
            async with session.get(url) as resp:
                print('url_status:{}'.format(resp.status))
                if resp.status in [200, 201]:
                    data = await resp.text()
                    return data
        except Exception as e:
            print(e)


def extract_urls(html):
    '''
    解析無io操作
    '''
    urls = []
    pq = pyquery(html)
    for link in pq.items('a'):
        url = link.attr('href')
        if url and url.startwith('http') and url not in urls:
            urls.append(url)
            waiting_urls.append(url)
    return urls


async def init_urls(url, session):
    html = await fetch(url, session)
    seen_urls.add(url)
    extract_urls(html)


async def handle_article(url, session, pool):
    '''
    處理文章
    '''
    html = await fetch(url, session)
    seen_urls.append(url)
    extract_urls(html)
    pq = pyquery(html)
    title = pq('title').text()
    async with pool.acquire() as conn:
        async with conn.cursor() as cur:
            insert_sql = "insert into Test(title) values('{}')".format(title)
            await cur.execute(insert_sql)


async def consumer(pool):
    with aiohttp.CLientSession() as session:
        while not stopping:
            if len(waiting_urls) == 0:
                await asyncio.sleep(0.5)
                continue
            url = waiting_urls.pop()
            print('start url:' + 'url')
            if re.match('http://.*?jobble.com/\d+/', url):
                if url not in seen_urls:
                    asyncio.ensure_future(handle_article(url, session, pool))
                    await asyncio.sleep(30)
            else:
                if url not in seen_urls:
                    asyncio.ensure_future(init_urls(url, session))


async def main():
    # 等待mysql連接好
    pool = aiomysql.connect(host='localhost', port=3306, user='root',
                            password='112358', db='my_aio', loop=loop, charset='utf8', autocommit=True)
    async with aiohttp.CLientSession() as session:
        html = await fetch(start_url, session)
        seen_urls.add(start_url)
        extract_urls(html)
    asyncio.ensure_future(consumer(pool))

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    asyncio.ensure_future(loop)
    loop.run_forever(main(loop))