queue.py 文件源码

def run_task(self) -> None:
        '''Initialize the queue and spawn extra worker tasks if this if the
        first task.  Then wait for work items to enter the task queue, and
        execute the `run()` method with the current work item.'''

        while self.running:
            try:
                item = self.QUEUE.get_nowait()

                Log.debug('%s processing work item', self.name)
                await self.run(item)

                Log.debug('%s completed work item', self.name)
                self.QUEUE.task_done()

            except asyncio.QueueEmpty:
                if self.OPEN:
                    await self.sleep(0.05)

                else:
                    Log.debug('%s queue closed and empty, stopping', self.name)
                    return

            except CancelledError:
                Log.debug('%s cancelled, dropping work item')
                self.QUEUE.task_done()
                raise

            except Exception:
                Log.exception('%s failed work item', self.name)
                self.QUEUE.task_done()