但是還有一點(diǎn)困擾我的是，當(dāng)我在后臺(tái)并發(fā)處理成百上千個(gè)文件或者執(zhí)行成百上千個(gè)進(jìn)程時(shí)，我總是擔(dān)心會(huì)不會(huì)有幾個(gè)任務(wù)偷偷掛了，整個(gè)代碼永遠(yuǎn)跑不完。我也很難知道代碼現(xiàn)在在哪里執(zhí)行。

最糟糕的是，當(dāng)我看著一個(gè)空白屏幕時(shí)，很難說(shuō)出我的代碼需要多長(zhǎng)時(shí)間才能執(zhí)行或 ETA 是多少。這對(duì)我安排工作日程的能力非常不利。

因此，我想要一種方法讓我知道代碼執(zhí)行到了哪里。

已有方法

比較傳統(tǒng)的做法是任務(wù)之間共享一塊內(nèi)存區(qū)域，在這塊內(nèi)存區(qū)域放一個(gè)計(jì)數(shù)器，當(dāng)一個(gè)任務(wù)結(jié)束的時(shí)候讓這個(gè)計(jì)數(shù)器+1，然后用一個(gè)線程不停的打印這個(gè)計(jì)數(shù)器的值。

這從來(lái)都不是一個(gè)好的解決方案：一方面，我需要在你現(xiàn)有的業(yè)務(wù)邏輯中添加一段用于計(jì)數(shù)的代碼，這違反了“低耦合，高內(nèi)聚”的原則。另一方面，由于線程安全問(wèn)題，我必須非常小心鎖定機(jī)制，這會(huì)導(dǎo)致不必要的性能問(wèn)題。

tqdm

有一天，我發(fā)現(xiàn)了 tqdm 庫(kù)，它使用進(jìn)度條來(lái)可視化我的代碼進(jìn)度。我可以使用進(jìn)度條來(lái)可視化我的 asyncio 任務(wù)的完成和預(yù)計(jì)到達(dá)時(shí)間嗎？

那么本文我把這個(gè)方法分享給大家，讓每個(gè)程序員都有機(jī)會(huì)監(jiān)控自己并發(fā)任務(wù)的進(jìn)度。

異步

在我們開(kāi)始之前，我希望您了解一些 Python asyncio 的背景知識(shí)。我的文章描述了asyncio的一些常用API的用法，這將有助于我們更好地理解tqdm的設(shè)計(jì)：

tqdm 概述

如官方網(wǎng)站所述，tqdm 是一個(gè)顯示循環(huán)進(jìn)度條的工具。它使用簡(jiǎn)單、高度可定制并且占用資源少。

一個(gè)典型的用法是將一個(gè)可迭代對(duì)象傳遞給 tqdm 構(gòu)造函數(shù)，然后你會(huì)得到一個(gè)如下所示的進(jìn)度條：

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

或者您可以在讀取文件時(shí)手動(dòng)瀏覽并更新進(jìn)度條的進(jìn)度：

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 上需要更多關(guān)于將 tqdm 與 asyncio 集成的信息。所以我深入研究了源代碼，看看 tqdm 是否支持 asyncio。

幸運(yùn)的是，最新版本的 tqdm 提供了包 tqdm.asyncio，它提供了類 tqdm_asyncio。

tqdm_asyncio 類有兩個(gè)相關(guān)的方法。一個(gè)是 tqdm_asyncio.as_completed。從源碼可以看出，它是對(duì)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)

另一個(gè)是 tqdm_asyncio.gather ，從源代碼可以看出，它基于模擬 asyncio.gather 功能的 tqdm_asyncio.as_completed 的實(shí)現(xiàn)：

@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)]

所以，接下來(lái)，我將描述這兩個(gè)API的用法。在開(kāi)始之前，我們還需要做一些準(zhǔn)備工作。在這里，我寫(xiě)了一個(gè)簡(jiǎn)單的方法來(lái)模擬一個(gè)隨機(jī)休眠時(shí)間的并發(fā)任務(wù)：

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

緊接著，我們將創(chuàng)建 2000 個(gè)并發(fā)任務(wù)，然后使用 tqdm_asyncio.gather 而不是熟悉的 asyncio.gather 方法來(lái)查看進(jìn)度條是否正常工作：

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