關於asyncio知識（四）

1、使用 asyncio 總結

最近在公司的一些項目中開始慢慢使用python 的asyncio, 使用的過程當中也是各類踩坑，遇到的問題也很多，其中有一次是內存的問題，本身也整理了遇到的問題以及解決方法詳細內容看：https://www.syncd.cn/article/memory_trouble

在前面整理的三篇asyncio文章中，也都是使用asyncio的一些方法，可是在實際項目中使用仍是避免不了碰到問題， 在這周的工做中遇到以前遇見過的問題，一個初學asyncio寫代碼中常常會碰到的問題，個人業務代碼在運行一段時間後提示以下錯誤提示：

Task was destroyed but it is pending!task: <Task pending coro=<HandleMsg.get_msg() done, defined at ex10.py:17> wait_for=<Future cancelled>>

這個錯誤我在前面幾篇關於asyncio的系列文章中也反覆說過這個問題，我也認爲本身不會在出現這種問題，可是意外的是，個人程序仍是出現了這個錯誤。

我將個人業務代碼經過一個demo代碼進行模擬復現以及解決這個問題，下面整理的就是這個過程

2、「Task was destroyed but it is pending!」

我經過下面這張圖先描述一下demo程序的邏輯：

 

 

import asyncio
from asyncio import Queue
import uuid
from asyncio import Lock
from asyncio import CancelledError
queue = Queue()
class HandleMsg(object):
    def __init__(self, unid, coroutine_queue, handle_manager):
        self.unid = unid
        self.coroutine_queue = coroutine_queue
        self.handle_manager = handle_manager
    async def get_msg(self):
        while True:
            coroutine_msg = await self.coroutine_queue.get()
            msg_type = coroutine_msg.get("msg")
            if msg_type == "start":
                print("recv unid [%s] is start" % self.unid)
            else:
                print("recv unid [%s] is end" % self.unid)
                # 每一個當一個unid收到end消息爲結束
                await self.handle_manager.del_unid(self.unid)
class HandleManager(object):
    """
    用於unid和queue的關係的處理
    """
    def __init__(self):
        self.loop = asyncio.get_event_loop()
        self.lock = Lock(loop=self.loop)
        self.handle_dict = dict()
    async def unid_bind(self, unid, coroutine_queue):
        async with self.lock:
            self.handle_dict[unid] = coroutine_queue
    async def get_queue(self, unid):
        async with self.lock:
            if unid in self.handle_dict:
                return self.handle_dict[unid]
    async def del_unid(self, unid):
        async with self.lock:
            if unid in self.handle_dict:
                self.handle_dict.pop(unid)
def make_uniqueid():
    """
    生成unid
    """
    uniqueid = str(uuid.uuid1())
    uniqueid = uniqueid.split("-")
    uniqueid.reverse()
    uniqueid = "".join(uniqueid)
    return uniqueid
async def product_msg():
    """
    生產者
    """
    while True:
        unid = make_uniqueid()
        msg_start = {"unid": unid, "msg": "start"}
        await queue.put(msg_start)
        msg_end = {"unid": unid, "msg": "end"}
        await queue.put(msg_end)
        loop = asyncio.get_event_loop()
        await asyncio.sleep(0.2, loop=loop)
async def consumer_from_queue(handle_manager):
    """
    消費者
    """
    while True:
        msg = await queue.get()
        print("consumer recv %s" % msg)
        msg_type = msg.get("msg")
        unid = msg.get("unid")
        if msg_type == "start":
            coroutine_queue = Queue()  # 用於和handle_msg協程進行數據傳遞
            handle_msg = HandleMsg(unid, coroutine_queue, handle_manager)
            await handle_manager.unid_bind(unid, coroutine_queue)
            await coroutine_queue.put(msg)
            loop = asyncio.get_event_loop()
            # 每次的start消息建立一個task 去處理消息
            loop.create_task(handle_msg.get_msg())
        else:
            coroutine_queue = await handle_manager.get_queue(unid)
            await coroutine_queue.put(msg)
if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    handle_manager = HandleManager()
    #  在最開始建立了兩個task 分別是生產者和消費者
    loop.create_task(product_msg())
    loop.create_task(consumer_from_queue(handle_manager))
    loop.run_forever()

上面的代碼表面上看沒啥問題，咱們先看看運行效果：

consumer recv {'unid': '784f436cfaf388f611e94ca974e1ffbe', 'msg': 'start'}
consumer recv {'unid': '784f436cfaf388f611e94ca974e1ffbe', 'msg': 'end'}
Task was destroyed but it is pending!
task: <Task pending coro=<HandleMsg.get_msg() done, defined at demo.py:17> wait_for=<Future cancelled>>
Task was destroyed but it is pending!
task: <Task pending coro=<HandleMsg.get_msg() done, defined at demo.py:17> wait_for=<Future cancelled>>
Task was destroyed but it is pending!
task: <Task pending coro=<HandleMsg.get_msg() done, defined at demo.py:17> wait_for=<Future cancelled>>
Task was destroyed but it is pending!
task: <Task pending coro=<HandleMsg.get_msg() done, defined at demo.py:17> wait_for=<Future cancelled>>
Task was destroyed but it is pending!
task: <Task pending coro=<HandleMsg.get_msg() done, defined at demo.py:17> wait_for=<Future cancelled>>
Task was destroyed but it is pending!
task: <Task pending coro=<HandleMsg.get_msg() done, defined at demo.py:17> wait_for=<Future cancelled>>
..........

程序沒運行一段時間都會出現上面顯示的錯誤提示，我先看看錯誤提示的信息：

 

1. Task was destroyed but it is pending!
2. task: <Task pending coro=<HandleMsg.get_msg() done, defined at demo.py:17> wait_for=<Future cancelled>>

 

上面提示的其實就是個人task 是在pendding狀態的時候被destroyed了，代碼行數以及調用方法都告訴咱們了是在：HandleMsg.get_msg() done, defined at demo.py:17

其實問題也比較好找，咱們爲每一個unid建立了一個task來處理消息，可是當咱們收到每一個unid消息的end消息以後其實這個task任務對於咱們來講就已經完成了，同時咱們刪除了個人unid和queue的綁定，可是咱們並無手動去取消這個task。

 
注意：這裏我其實也有一個不理解的地方：關於這個task爲何會會destroyed，這個協程裏是一個死循環一直在收消息，當queue裏面沒有消息協程也應該一直在await 地方在等待纔對，可是若是咱們把收到end消息的那個地方的刪除unid和queue的綁定關係不刪除，那麼這個任務是不會被descroyed。因此沒有徹底明白這裏的機制，若是明白的同窗歡迎留言討論

 

可是即便上面的機制咱們有點不是特別明白，咱們其實也應該把這個task手動進行cancel的，咱們們將上面的代碼稍微進行改動以下：

async def get_msg(self):
        try:
            while True:
                coroutine_msg = await self.coroutine_queue.get()
                msg_type = coroutine_msg.get("msg")
                if msg_type == "start":
                    print("recv unid [%s] is start" % self.unid)
                else:
                    print("recv unid [%s] is end" % self.unid)
                    # 每一個當一個unid收到end消息爲結束
                    await self.handle_manager.del_unid(self.unid)
                    current_task = asyncio.Task.current_task()
                    current_task.cancel()   # 手動cancel 當前的當前的task
        except CancelledError as e:
            print("unid [%s] cancelled success" %self.unid)

這裏有個問題須要注意就是當咱們對task進行cancel的時候會拋出cancelledError異常，咱們須要對異常進行處理。官網也對此進行專門說明：
https://docs.python.org/3.6/library/asyncio-task.html#coroutine

內容以下：

cancel()
Request that this task cancel itself.
This arranges for a CancelledError to be thrown into the wrapped coroutine on the next cycle through the event loop. The coroutine then has a chance to clean up or even deny the request using try/except/finally.
Unlike Future.cancel(), this does not guarantee that the task will be cancelled: the exception might be caught and acted upon, delaying cancellation of the task or preventing cancellation completely. The task may also return a value or raise a different exception.
Immediately after this method is called, cancelled() will not return True (unless the task was already cancelled). A task will be marked as cancelled when the wrapped coroutine terminates with a CancelledError exception (even if cancel() was not called).

3、小結

雖然還有一些地方不太明白，可是隨着用的越多，碰到的問題越多，一個一個解決，可能如今對某些知識還有點模糊，可是至少比剛開始使用asyncio的時候清晰了好多，以前整理的三篇文章的鏈接以下：
https://www.syncd.cn/article/asyncio_article_01
https://www.syncd.cn/article/asyncio_article_02
https://www.syncd.cn/article/asyncio_article_03

也歡迎加入交流羣一塊兒討論相關內容：948510543