关于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>>

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>>
..........

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)

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).