workers.py 文件源码

def worker(self, id_):
        loop = asyncio.get_event_loop()
        asyncio.Task.current_task().logging_id = 'wk%02d' % (id_,)

        while True:
            logger.debug("Worker %d waiting.", id_)
            item = yield from self.queue.get()
            logger.debug(
                "Worker %d working on %s %s.",
                id_, item.__class__.__name__, item,
            )
            task = loop.create_task(item())
            try:
                res = yield from task
                item.set_result(res)
            except CancelledError:
                item.cancel()
                logger.warn("Cancel of %s", item)
            except Exception as e:
                item.set_exception(e)
                if SETTINGS.VERBOSE or SETTINGS.DEBUG:
                    logger.exception("Failed to process %s: %s", item, e)
                else:
                    logger.error("Failed to process %s: %s", item, e)
            finally:
                self.queue.task_done()