乐趣区

关于程序员:在-Python-中将-Tqdm-与-Asyncio-结合使用

jiezi

动动发财的小手，点个赞吧！

简介

困扰

在 Python 中应用并发编程来提高效率对于数据科学家来说并不常见。在后盾察看各种子过程或并发线程以放弃我的计算或 IO 绑定工作的程序总是令人满意的。

然而还有一点困扰我的是，当我在后盾并发解决成千盈百个文件或者执行成千盈百个过程时，我总是放心会不会有几个工作偷偷挂了，整个代码永远跑不完。我也很难晓得代码当初在哪里执行。

最蹩脚的是，当我看着一个空白屏幕时，很难说出我的代码须要多长时间能力执行或 ETA 是多少。这对我安顿工作日程的能力十分不利。

因而，我想要一种办法让我晓得代码执行到了哪里。

已有办法

比拟传统的做法是工作之间共享一块内存区域，在这块内存区域放一个计数器，当一个工作完结的时候让这个计数器 +1，而后用一个线程不停的打印这个计数器的值。

这素来都不是一个好的解决方案：一方面，我须要在你现有的业务逻辑中增加一段用于计数的代码，这违反了“低耦合，高内聚”的准则。另一方面，因为线程平安问题，我必须十分小心锁定机制，这会导致不必要的性能问题。

tqdm

有一天，我发现了 tqdm 库，它应用进度条来可视化我的代码进度。我能够应用进度条来可视化我的 asyncio 工作的实现和预计达到工夫吗？

那么本文我把这个办法分享给大家，让每个程序员都有机会监控本人并发工作的进度。

异步

在咱们开始之前，我心愿您理解一些 Python asyncio 的背景常识。我的文章形容了 asyncio 的一些罕用 API 的用法，这将有助于咱们更好地了解 tqdm 的设计：

tqdm 概述

如官方网站所述，tqdm 是一个显示循环进度条的工具。它应用简略、高度可定制并且占用资源少。

一个典型的用法是将一个可迭代对象传递给 tqdm 构造函数，而后你会失去一个如下所示的进度条：

from time import sleep
from tqdm import tqdm


def main():
    for _ in tqdm(range(100)):
        # do something in the loop
        sleep(0.1)


if __name__ == "__main__":
    main()

或者您能够在读取文件时手动浏览并更新进度条的进度：

import os
from tqdm import tqdm


def main():
    filename = "../data/large-dataset"
    with (tqdm(total=os.path.getsize(filename)) as bar,
            open(filename, "r", encoding="utf-8") as f):
        for line in f:
            bar.update(len(line))


if __name__ == "__main__":
    main()

将 tqdm 与异步集成

总体而言，tqdm 十分易于应用。然而，GitHub 上须要更多对于将 tqdm 与 asyncio 集成的信息。所以我深入研究了源代码，看看 tqdm 是否反对 asyncio。

侥幸的是，最新版本的 tqdm 提供了包 tqdm.asyncio，它提供了类 tqdm_asyncio。

tqdm_asyncio 类有两个相干的办法。一个是 tqdm_asyncio.as_completed。从源码能够看出，它是对 asyncio.as_completed 的包装：

@classmethod
    def as_completed(cls, fs, *, loop=None, timeout=None, total=None, **tqdm_kwargs):
        """Wrapper for `asyncio.as_completed`."""
        if total is None:
            total = len(fs)
        kwargs = {}
        if version_info[:2] < (3, 10):
            kwargs['loop'] = loop
        yield from cls(asyncio.as_completed(fs, timeout=timeout, **kwargs),
                       total=total, **tqdm_kwargs)

另一个是 tqdm_asyncio.gather，从源代码能够看出，它基于模仿 asyncio.gather 性能的 tqdm_asyncio.as_completed 的实现：

@classmethod
    async def gather(cls, *fs, loop=None, timeout=None, total=None, **tqdm_kwargs):
        """Wrapper for `asyncio.gather`."""
        async def wrap_awaitable(i, f):
            return i, await f

        ifs = [wrap_awaitable(i, f) for i, f in enumerate(fs)]
        res = [await f for f in cls.as_completed(ifs, loop=loop, timeout=timeout,
                                                 total=total, **tqdm_kwargs)]
        return [i for _, i in sorted(res)]

所以，接下来，我将形容这两个 API 的用法。在开始之前，咱们还须要做一些筹备工作。在这里，我写了一个简略的办法来模仿一个随机休眠工夫的并发工作：

import asyncio
import random

from tqdm.asyncio import tqdm_asyncio


class AsyncException(Exception):
    def __int__(self, message):
        super.__init__(self, message)


async def some_coro(simu_exception=False):
    delay = round(random.uniform(1.0, 5.0), 2)

    # We will simulate throwing an exception if simu_exception is True
    if delay > 4 and simu_exception:
        raise AsyncException("something wrong!")

    await asyncio.sleep(delay)

    return delay

紧接着，咱们将创立 2000 个并发工作，而后应用 tqdm_asyncio.gather 而不是相熟的 asyncio.gather 办法来查看进度条是否失常工作：

async def main():
    tasks = []
    for _ in range(2000):
        tasks.append(some_coro())
    await tqdm_asyncio.gather(*tasks)

    print(f"All tasks done.")


if __name__ == "__main__":
    asyncio.run(main())

或者让咱们用 tqdm_asyncio.as_completed 替换 tqdm_asyncio.gather 并重试：

async def main():
    tasks = []
    for _ in range(2000):
        tasks.append(some_coro())

    for done in tqdm_asyncio.as_completed(tasks):
        await done

    print(f"The tqdm_asyncio.as_completed also works fine.")


if __name__ == "__main__":
    asyncio.run(main())

本文由 mdnice 多平台公布

退出移动版