技術標籤：軟體測試python

通過yield生成器來實現協程

import time

def consumer(): #函式中如果有yield，那麼c=consumer()不會執行該函式，只有執行了c.__next__()才會執行該函式
    print('消費者準備吃包子')
    while True:
        baozi=yield #yield生成器，用於接收其他函式中send傳送的資料，當接收到資料後才會執行下面的程式碼
        print('消費者吃了包子',baozi)

def producer():
    c=consumer() #該函式暫時不會執行，因為含有yield，只有當c.__next__()執行時，才會執行c=consumer()
 

    c.__next__() #consumer()函式中含有yield時，不會執行，只有執行了c.__next__()，才會執行consumer()
    for i in range(5):
        time.sleep(1)
        print('生產者生產了包子',i)
        c.send(i) #該函式將值傳送給yield，yield後面的程式碼才會執行
        
if __name__=='__main__':
    producer()

通過greenlet實現協程，手動切換任務執行

from greenlet import greenlet

def work1(
 
):
    print(1)
    gr2.switch() #暫停gr1,切換到gr2執行
    print(3)
    gr2.switch()
    print('任務1執行完畢')
    
def work2():
    print(2)
    gr1.switch() #暫停gr2,切換到gr1執行
    print(4)
    print('任務2執行完畢')
    gr1.switch()
    
if __name__=='__main__':
    gr1=greenlet(work1) #建立協程1
    gr2=greenlet(work2) #建立協程1
 

    gr1.switch() #gr1先執行
    print('主程式執行完畢')

自動切換任務執行

import gevent

def work1():
    print('任務1執行')
    gevent.sleep(3)
    print('任務1執行完畢')
    
def work2():
    print('任務2執行')
    gevent.sleep(2)
    print('任務2執行完畢')
    
def work3():
    print('任務3執行')
    gevent.sleep(1)
    print('任務3執行完畢')
    
if __name__=='__main__':
    gevent.joinall([
        gevent.spawn(work1), #work1阻塞時，自動切換到work2
        gevent.spawn(work2), #work2阻塞時，自動切換到work3
        gevent.spawn(work3),
        ])

通過協程載入多個網頁

from gevent import monkey; monkey.patch_all() #此處必須新增此函式，保證多協程併發
import gevent
import requests

def f(url):
    print('開始請求網頁：',url)
    resp=requests.get(url)
    print(resp.text[:100])
    print('結束：',url)
    print('-----------------------------------------------------')
    
if __name__=='__main__':
    gevent.joinall([
    gevent.spawn(f,'https://www.baidu.com'),
    gevent.spawn(f,'https://www.163.com/'),
    gevent.spawn(f,'https://www.sina.com.cn/'),
    ])

asyncio用法介紹

import time
import asyncio

async def work(): #通過async關鍵字，將此函式定義為協程的任務函式
    print('任務執行了')
    
if __name__=='__main__':
    coroutine=work() #建立任務函式物件，由於async關鍵字存在，此時函式還未執行
    loop=asyncio.get_event_loop() #建立一個事件迴圈的協程物件
    loop.run_until_complete(coroutine) #將任務函式傳入事件迴圈，執行任務函式

asyncio多協程用法

import time
import asyncio

async def work1():
    print('任務1執行了')
    return 1
    
async def work2():
    print('任務2執行了')
    return 2
    
if __name__=='__main__':
    coroutine1=work1()
    coroutine2=work2()
    loop=asyncio.get_event_loop()
    task1=asyncio.ensure_future(coroutine1) #該函式可獲取任務函式的返回值
    task2=asyncio.ensure_future(coroutine2)
    loop.run_until_complete(task1)
    loop.run_until_complete(task2)
    print('任務1的返回值：',task1.result()) #列印任務函式的返回值
    print('任務2的返回值：',task2.result())