第十二天 異步編程(2)和單元測(cè)試

今天計(jì)劃學(xué)習(xí)Python的多線(xiàn)程編程異步編程，學(xué)習(xí)項(xiàng)目及練習(xí)源碼地址：
GitHub源碼

協(xié)程

參見(jiàn)昨天的學(xué)習(xí)記錄

無(wú)阻塞

異步程序依然會(huì)假死freezing

freezing案例：

import asyncio
import time
import threading

#定義一個(gè)異步操作
async def hello1(a,b):
    print(f"異步函數(shù)開(kāi)始執(zhí)行")
    await asyncio.sleep(3)
    print("異步函數(shù)執(zhí)行結(jié)束")
    return a+b

#在一個(gè)異步操作里面調(diào)用另一個(gè)異步操作
async def main():
    c=await hello1(10,20)
    print(c)
    print("主函數(shù)執(zhí)行")

loop = asyncio.get_event_loop()
tasks = [main()]
loop.run_until_complete(asyncio.wait(tasks))

loop.close()

'''運(yùn)行結(jié)果為：
異步函數(shù)開(kāi)始執(zhí)行（在此處要等待3秒）
異步函數(shù)執(zhí)行結(jié)束
30
主函數(shù)執(zhí)行
'''

例子中岛宦，hello1是一個(gè)耗時(shí)3s的異步任務(wù)惶楼，main也是一個(gè)異步方法狰腌，但是main需要調(diào)用hello1的返回值您宪，所以必須登臺(tái)hello1執(zhí)行完成才能繼續(xù)執(zhí)行main痕惋，這說(shuō)明異步也是會(huì)有阻塞的。

而之前定義的異步函數(shù)不用等待是因?yàn)槭录h(huán)將所有的異步操作‘gather’起來(lái)蚕键，在多個(gè)操作間不同的游走切換，來(lái)回調(diào)用所有沒(méi)有等待衰粹。

也可以理解為锣光，事件循環(huán)只有一個(gè)異步操作在處理，沒(méi)有可以切換執(zhí)行的目標(biāo)铝耻，所以只能等待當(dāng)前的操作完成誊爹。

多線(xiàn)程+asyncio解決調(diào)用時(shí)freezing

為了讓一個(gè)協(xié)程函數(shù)在不同的線(xiàn)程中執(zhí)行，我們可以使用以下兩個(gè)函數(shù):

1. loop.call_soon_threadsafe(callback, *args)瓢捉，這是一個(gè)很底層的API接口替废，一般很少使用
2. asyncio.run_coroutine_threadsafe(coroutine，loop) 第一個(gè)參數(shù)為需要異步執(zhí)行的協(xié)程函數(shù)泊柬，第二個(gè)loop參數(shù)為在新線(xiàn)程中創(chuàng)建的事件循環(huán)loop，注意一定要是在新線(xiàn)程中創(chuàng)建哦诈火，該函數(shù)的返回值是一個(gè)concurrent.futures.Future類(lèi)的對(duì)象兽赁，用來(lái)獲取協(xié)程的返回結(jié)果。 future = asyncio.run_coroutine_threadsafe(coro_func(), loop) 在新線(xiàn)程中運(yùn)行協(xié)程result = future.result()等待獲取Future的結(jié)果

示例代碼：

import asyncio 

import asyncio,time,threading

#需要執(zhí)行的耗時(shí)異步任務(wù)
async def func(num):
    print(f'準(zhǔn)備調(diào)用func,大約耗時(shí){num}')
    await asyncio.sleep(num)
    print(f'耗時(shí){num}之后,func函數(shù)運(yùn)行結(jié)束')

#定義一個(gè)專(zhuān)門(mén)創(chuàng)建事件循環(huán)loop的函數(shù)冷守，在另一個(gè)線(xiàn)程中啟動(dòng)它
def start_loop(loop):
    asyncio.set_event_loop(loop)
    loop.run_forever()

#定義一個(gè)main函數(shù)
def main():
    coroutine1 = func(3)
    coroutine2 = func(2)
    coroutine3 = func(1)

    new_loop = asyncio.new_event_loop()                        #在當(dāng)前線(xiàn)程下創(chuàng)建時(shí)間循環(huán)刀崖，（未啟用），在start_loop里面啟動(dòng)它
    t = threading.Thread(target=start_loop,args=(new_loop,))   #通過(guò)當(dāng)前線(xiàn)程開(kāi)啟新的線(xiàn)程去啟動(dòng)事件循環(huán)
    t.start()

    asyncio.run_coroutine_threadsafe(coroutine1,new_loop)  #這幾個(gè)是關(guān)鍵拍摇，代表在新線(xiàn)程中事件循環(huán)不斷“游走”執(zhí)行
    asyncio.run_coroutine_threadsafe(coroutine2,new_loop)
    asyncio.run_coroutine_threadsafe(coroutine3,new_loop)

    for i in "iloveu":
        print(str(i)+"    ")

if __name__ == "__main__":
    main()

'''運(yùn)行結(jié)果為：
i    準(zhǔn)備調(diào)用func,大約耗時(shí)3
l    準(zhǔn)備調(diào)用func,大約耗時(shí)2
o    準(zhǔn)備調(diào)用func,大約耗時(shí)1
v
e
u
耗時(shí)1之后,func函數(shù)運(yùn)行結(jié)束
耗時(shí)2之后,func函數(shù)運(yùn)行結(jié)束
耗時(shí)3之后,func函數(shù)運(yùn)行結(jié)束
'''

第一步：定義需要異步執(zhí)行的一系列操作亮钦，及一系列協(xié)程函數(shù)；

第二步：在主線(xiàn)程中定義一個(gè)新的線(xiàn)程充活，然后在新線(xiàn)程中產(chǎn)生一個(gè)新的事件循環(huán)蜂莉；

第三步：在主線(xiàn)程中，通過(guò)asyncio.run_coroutine_threadsafe(coroutine,loop)這個(gè)方法混卵，將一系列異步方法注冊(cè)到新線(xiàn)程的loop里面去映穗，這樣就是新線(xiàn)程負(fù)責(zé)事件循環(huán)的執(zhí)行。

使用asyncio實(shí)現(xiàn)一個(gè)timer 定時(shí)器

所謂的timer指的是幕随，指定一個(gè)時(shí)間間隔蚁滋，讓某一個(gè)操作隔一個(gè)時(shí)間間隔執(zhí)行一次，如此周而復(fù)始赘淮。很多編程語(yǔ)言都提供了專(zhuān)門(mén)的timer實(shí)現(xiàn)機(jī)制辕录、包括C++、C#等梢卸。但是 Python 并沒(méi)有原生支持 timer走诞，不過(guò)可以用 asyncio.sleep 模擬。大致的思想如下低剔，將timer定義為一個(gè)異步協(xié)程速梗，然后通過(guò)事件循環(huán)去調(diào)用這個(gè)異步協(xié)程肮塞，讓事件循環(huán)不斷在這個(gè)協(xié)程中反反復(fù)調(diào)用，只不過(guò)隔幾秒調(diào)用一次即可姻锁。簡(jiǎn)單的實(shí)現(xiàn)如下（本例基于python3.7:

import asyncio
async def delay(time):
    await asyncio.sleep(time)

async def timer(time,function):
    while True:
        future=asyncio.ensure_future(delay(time))
        await future
        future.add_done_callback(function)

def func(future):
    print('done')

if __name__=='__main__':
    asyncio.run(timer(2,func))

aiohttp模塊

asyncio可以實(shí)現(xiàn)單線(xiàn)程并發(fā)IO操作枕赵。如果僅用在客戶(hù)端，發(fā)揮的威力不大位隶。如果把a(bǔ)syncio用在服務(wù)器端拷窜，例如Web服務(wù)器，由于HTTP連接就是IO操作涧黄，因此可以用單線(xiàn)程+coroutine實(shí)現(xiàn)多用戶(hù)的高并發(fā)支持篮昧。

asyncio實(shí)現(xiàn)了TCP、UDP笋妥、SSL等協(xié)議懊昨，aiohttp則是基于asyncio實(shí)現(xiàn)的HTTP框架。

• 安裝

  pip3 install aiohttp

• 示例代碼

import asyncio

from aiohttp import web

async def index(request):
    await asyncio.sleep(0.5)
    return web.Response(body=b'<h1>Index</h1>')

async def hello(request):
    await asyncio.sleep(0.5)
    text = '<h1>hello, {}!</h1>'.format(request.match_info['name'])
    return web.Response(body=text.encode('utf-8'))

async def init(loop):
    app = web.Application(loop=loop)
    app.router.add_route('GET', '/', index)
    app.router.add_route('GET', '/hello/{name}', hello)
    srv = await loop.create_server(app.make_handler(), '127.0.0.1', 8000)
    print('Server started at http://127.0.0.1:8000...')
    return srv

loop = asyncio.get_event_loop()
loop.run_until_complete(init(loop))
loop.run_forever()

aiomysql

Python3.7+ 下的一個(gè)異步操作mysql數(shù)據(jù)的模塊,官方地址

示例：

#coding:utf-8

import aiomysql
import asyncio
import logging
import traceback
'''
mysql 異步版本
'''

logobj = logging.getLogger('mysql')

class Pmysql:
    __connection = None

    def __init__(self):
        self.cursor = None
        self.connection = None

    @staticmethod
    async def getconnection():
        if Pmysql.__connection == None:
            conn = await aiomysql.connect(
                host='127.0.0.1',
                port=3306,
                user='root',
                password='123456',
                db='mytest',
                )
            if conn:
                Pmysql.__connection = conn
                return conn
            else:
                raise("connect to mysql error ")
        else:
            return Pmysql.__connection

    async def query(self,query,args=None):
        self.cursor = await self.connection.cursor()
        await self.cursor.execute(query,args)
        r = await self.cursor.fetchall()
        await self.cursor.close()
        return r


async def test():
    conn = await Pmysql.getconnection()
    mysqlobj.connection = conn
    await conn.ping()
    r = await mysqlobj.query("select * from person")
    for i in r:
        print(i)
    conn.close()

if __name__ == '__main__':
    mysqlobj = Pmysql()
    loop = asyncio.get_event_loop()
    loop.run_until_complete(test())

aioredis

redis異步操作庫(kù)春宣，官方地址

示例：

import aioredis
import asyncio

class Redis:
    _redis = None

    async def get_redis_pool(self, *args, **kwargs):
        if not self._redis:
            self._redis = await aioredis.create_redis_pool(*args, **kwargs)
        return self._redis

    async def close(self):
        if self._redis:
            self._redis.close()
            await self._redis.wait_closed()


async def get_value(key):
    redis = Redis()
    r = await redis.get_redis_pool(('127.0.0.1', 6379), db=7, encoding='utf-8')
    value = await r.get(key)
    print(f'{key!r}: {value!r}')
    await redis.close()         

if __name__ == '__main__':
    asyncio.run(get_value('key'))  # need python3.7

測(cè)試

單元測(cè)試

單元測(cè)試是用來(lái)對(duì)一個(gè)模塊酵颁、一個(gè)函數(shù)或者一個(gè)類(lèi)來(lái)進(jìn)行正確性檢驗(yàn)的測(cè)試工作。

Python自帶的unittest模塊可以很方便的讓我們編寫(xiě)單元測(cè)試月帝。

編寫(xiě)單元測(cè)試時(shí)躏惋，我們需要編寫(xiě)一個(gè)測(cè)試類(lèi)，從unittest.TestCase繼承嚷辅。
以test開(kāi)頭的方法就是測(cè)試方法簿姨，不以test開(kāi)頭的方法不被認(rèn)為是測(cè)試方法，測(cè)試的時(shí)候不會(huì)被執(zhí)行簸搞。

對(duì)每一類(lèi)測(cè)試都需要編寫(xiě)一個(gè)test_xxx()方法扁位。由于unittest.TestCase提供了很多內(nèi)置的條件判斷，我們只需要調(diào)用這些方法就可以斷言輸出是否是我們所期望的攘乒。

代碼示例:

'''
定義一個(gè)要測(cè)試的類(lèi)
mydict.py
'''
class MyDict(dict):

    def __init__(self, **kw):
        super().__init__(**kw)

    def __getattr__(self, key):
        try:
            return self[key]
        except KeyError:
            raise AttributeError(r"'Dict' object has no attribute '%s'" % key)

    def __setattr__(self, key, value):
        self[key] = value

編寫(xiě)單元測(cè)試：

import unittest

from mydict import MyDict

class TestDict(unittest.TestCase):

    def test_init(self):
        d = MyDict(a=1, b='test')
        self.assertEqual(d.a, 1)
        self.assertEqual(d.b, 'test')
        self.assertTrue(isinstance(d, dict))

    def test_key(self):
        d = MyDict()
        d['key'] = 'value'
        self.assertEqual(d.key, 'value')

    def test_attr(self):
        d = MyDict()
        d.key = 'value'
        self.assertTrue('key' in d)
        self.assertEqual(d['key'], 'value')

    def test_keyerror(self):
        d = MyDict()
        with self.assertRaises(KeyError):
            value = d['empty']

    def test_attrerror(self):
        d = MyDict()
        with self.assertRaises(AttributeError):
            value = d.empty

運(yùn)行單元測(cè)試

一旦編寫(xiě)好單元測(cè)試贤牛，我們就可以運(yùn)行單元測(cè)試。最簡(jiǎn)單的運(yùn)行方式是在mydict_test.py的最后加上兩行代碼：

if __name__ == '__main__':
    unittest.main()

另一種方法是在命令行通過(guò)參數(shù)-m unittest直接運(yùn)行單元測(cè)試：

python -m unittest mydict_test

這是推薦的做法则酝，因?yàn)檫@樣可以一次批量運(yùn)行很多單元測(cè)試殉簸，并且，有很多工具可以自動(dòng)來(lái)運(yùn)行這些單元測(cè)試沽讹。

setUp() tearDown()在每次執(zhí)行之前準(zhǔn)備環(huán)境般卑，或者在每次執(zhí)行完之后需要進(jìn)行一些清理。比如執(zhí)行前需要連接數(shù)據(jù)庫(kù)爽雄，執(zhí)行完成之后需要還原數(shù)據(jù)蝠检、斷開(kāi)連接。

如果想要在所有case執(zhí)行之前準(zhǔn)備一次環(huán)境挚瘟，并在所有case執(zhí)行結(jié)束之后再清理環(huán)境叹谁，我們可以用 setUpClass() 與 tearDownClass()

跳過(guò)某個(gè)case需要用到skip裝飾器一共有三個(gè)：unittest.skip(reason)饲梭、unittest.skipIf(condition, reason)、unittest.skipUnless(condition, reason)焰檩，skip無(wú)條件跳過(guò)憔涉，skipIf當(dāng)condition為T(mén)rue時(shí)跳過(guò)，skipUnless當(dāng)condition為False時(shí)跳過(guò)析苫。

在VS Code中對(duì)Python進(jìn)行單元測(cè)試

Python擴(kuò)展支持使用Python的內(nèi)置unittest框架以及pytest和Nose進(jìn)行單元測(cè)試兜叨。要使用pytest和Nose，必須將它們安裝到當(dāng)前的Python環(huán)境中（即衩侥，在pythonPath設(shè)置中標(biāo)識(shí)的環(huán)境国旷，請(qǐng)參閱環(huán)境）。

使用Python：Discover Unit Tests根據(jù)當(dāng)前所選測(cè)試框架的發(fā)現(xiàn)模式掃描項(xiàng)目以進(jìn)行測(cè)試（請(qǐng)參閱測(cè)試發(fā)現(xiàn)茫死。一旦發(fā)現(xiàn)跪但，VS Code提供了多種運(yùn)行測(cè)試的方法（請(qǐng)參閱運(yùn)行測(cè)試）。

單元測(cè)試輸出顯示在Python Test Log面板中峦萎，包括未安裝測(cè)試框架時(shí)導(dǎo)致的錯(cuò)誤特漩。

在settings.json中進(jìn)行設(shè)置：

{
    "python.pythonPath": "/usr/local/bin/python3",
    "python.testing.unittestEnabled": true,
    "python.testing.unittestArgs": [
        "-v",
        "-s",
        "./src/tests",
        "-p",
        "test_*.py"
    ],
    "python.testing.pytestEnabled": false,
    "python.testing.nosetestsEnabled": false,
}

Unittest配置設(shè)置

UnitTest的默認(rèn)參數(shù)如下：

-v設(shè)置默認(rèn)詳細(xì)程度雳灾。刪除此參數(shù)以獲得更簡(jiǎn)單的輸出。

-s .指定用于發(fā)現(xiàn)測(cè)試的起始目錄冯凹。如果您在“test”文件夾中進(jìn)行了測(cè)試谎亩，則可以將其更改為-s test（"-s", "test"在arguments數(shù)組中）。

-p test.py是用于查找測(cè)試的發(fā)現(xiàn)模式宇姚。在這種情況下匈庭，它.py是包含單詞“test” 的任何文件。如果以不同的方式命名測(cè)試文件浑劳，例如在每個(gè)文件名后附加“_test”阱持，則使用類(lèi)似于*_test.py數(shù)組的相應(yīng)參數(shù)的模式。

要在第一次失敗時(shí)停止測(cè)試運(yùn)??行魔熏，請(qǐng)將fail fast選項(xiàng)添加"-f"到arguments數(shù)組中衷咽。

文檔測(cè)試

如果你經(jīng)常閱讀Python的官方文檔鸽扁，可以看到很多文檔都有示例代碼。比如re模塊就帶了很多示例代碼：

>>> import re
>>> m = re.search('(?<=abc)def', 'abcdef')
>>> m.group(0)
'def'

可以把這些示例代碼在Python的交互式環(huán)境下輸入并執(zhí)行镶骗，結(jié)果與文檔中的示例代碼顯示的一致桶现。

這些代碼與其他說(shuō)明可以寫(xiě)在注釋中，然后卖词，由一些工具來(lái)自動(dòng)生成文檔巩那。既然這些代碼本身就可以粘貼出來(lái)直接運(yùn)行，那么此蜈，可不可以自動(dòng)執(zhí)行寫(xiě)在注釋中的這些代碼呢即横？

答案是肯定的。

當(dāng)我們編寫(xiě)注釋時(shí)裆赵，如果寫(xiě)上這樣的注釋?zhuān)?/p>

def abs(n):
    '''
    Function to get absolute value of number.
    
    Example:
    
    >>> abs(1)
    1
    >>> abs(-1)
    1
    >>> abs(0)
    0
    '''
    return n if n >= 0 else (-n)

無(wú)疑更明確地告訴函數(shù)的調(diào)用者該函數(shù)的期望輸入和輸出东囚。

并且，Python內(nèi)置的“文檔測(cè)試”（doctest）模塊可以直接提取注釋中的代碼并執(zhí)行測(cè)試战授。

doctest嚴(yán)格按照Python交互式命令行的輸入和輸出來(lái)判斷測(cè)試結(jié)果是否正確页藻。只有測(cè)試異常的時(shí)候，可以用...表示中間一大段煩人的輸出植兰。

小結(jié)

今天主要學(xué)習(xí)了Pyton的異步編程份帐，并簡(jiǎn)單了解了下相關(guān)的常用模塊。并針對(duì)單元測(cè)試進(jìn)行了詳細(xì)的了解楣导，單元測(cè)試很重要废境。明天打算開(kāi)始學(xué)習(xí)下Pyton的函數(shù)式編程。