關於asyncio知識（四）

一、使用 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代碼進行模擬復現以及解決這個問題，下面整理的就是這個過程

二、“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).

三、小結

雖然還有一些地方不太明白，但是隨着用的越多，碰到的問題越多，一個一個解決，可能現在對某些知識還有點模糊，但是至少比剛開始使用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