Python生成器和協(xié)程怎么用

本篇內(nèi)容主要講解“Python生成器和協(xié)程怎么用”，感興趣的朋友不妨來看看。本文介紹的方法操作簡(jiǎn)單快捷，實(shí)用性強(qiáng)。下面就讓小編來帶大家學(xué)習(xí)“Python生成器和協(xié)程怎么用”吧!

認(rèn)識(shí)生成器

你將如何生成任意長(zhǎng)度的斐波那契數(shù)列？顯然，你需要跟蹤一些數(shù)據(jù)，并且需要以某種方式對(duì)其進(jìn)行操作以創(chuàng)建下一個(gè)元素。

你的第一直覺可能是創(chuàng)建一個(gè)可迭代的類，這不失是一個(gè)好方法。讓我們開始，使用我們?cè)谇懊鎺坠?jié)中已經(jīng)介紹過的內(nèi)容：

class Fibonacci:

    def __init__(self, limit):
        self.n1 = 0
        self.n2 = 1
        self.n = 1
        self.i = 1
        self.limit = limit

    def __iter__(self):
        return self

    def __next__(self):
        if self.i > self.limit:
            raise StopIteration

        if self.i > 1:
            self.n = self.n1 + self.n2
            self.n1, self.n2 = self.n2, self.n

        self.i += 1
        return self.n


fib = Fibonacci(10)
for i in fib:
    print(i)

讓我們把它變得更緊湊。

如果你到目前為止一直在關(guān)注該系列，那么這里可能不會(huì)有任何驚喜。然而，對(duì)于像序列這樣簡(jiǎn)單的事情，這種方法可能會(huì)讓人覺得有點(diǎn)過頭了。

這種情況正是生成器的用途。

def fibonacci(limit):
    if limit >= 1:
        yield (n2 := 1)

    n1 = 0

    for _ in range(1, limit):
        yield (n := n1 + n2)
        n1, n2 = n2, n


for i in fibonacci(10):
    print(i)

生成器看起來肯定更緊湊——只有 9 行長(zhǎng)，而類為 22 行——但它同樣可讀。

關(guān)鍵是yield關(guān)鍵字，它返回一個(gè)值而不退出函數(shù)。yield在功能上與我們類中的__next__()函數(shù)相同。生成器將運(yùn)行到（并包括）它的yield語(yǔ)句，然后在它做任何事情之前等待另一個(gè)__next__()調(diào)用。一旦它得到那個(gè)調(diào)用，它將繼續(xù)運(yùn)行，直到它碰到另一個(gè)yield。

注意：看起來很奇怪的:=是 Python 3.8 中的新“海象運(yùn)算符”，它分配并返回一個(gè)值。如果你使用的是 Python 3.7 或更早版本，則可以將這些語(yǔ)句分成兩行（單獨(dú)去賦值和寫yield語(yǔ)句）。

你還會(huì)注意到缺少raise StopIteration聲明。生成器不需要它們；事實(shí)上，自PEP 479以來，他們甚至不允許他們這樣做。當(dāng)生成器函數(shù)自然終止或使用return語(yǔ)句終止時(shí)，StopIteration會(huì)在幕后自動(dòng)觸發(fā)。

嘗試生成器

修訂日期：2019 年 11 月 29 日

曾經(jīng)規(guī)定了yield不能出現(xiàn)在代碼中try子句中的try-finally中。PEP 255定義了生成器語(yǔ)法，解釋了原因：

難點(diǎn)在于不能保證生成器會(huì)被恢復(fù)，因此不能保證 finally 塊會(huì)被執(zhí)行；這就違背finally的目的了。

這在 PEP 342 PEP 342中進(jìn)行了更改，并在 Python 2.5 中完成。

那么，為什么要討論這樣一個(gè)古老的變化呢？簡(jiǎn)單：直到今天，我的印象是yield無(wú)法出現(xiàn)在try-finally中. 一些關(guān)于該主題的文章錯(cuò)誤地引用了舊規(guī)則。

把生成器作為對(duì)象

你可能還記得 Python 將函數(shù)視為對(duì)象，生成器也不例外！在我們之前的示例的基礎(chǔ)上，我們可以保存生成器的特定實(shí)例。

例如，如果我只想打印斐波那契數(shù)列的第 10-20 個(gè)值怎么辦？

首先，我將生成器保存在一個(gè)變量中，以便我可以重用它。限制對(duì)我來說并不重要，所以我會(huì)使用大的限制。使用我的循環(huán)范圍來更容易顯示內(nèi)容，因?yàn)檫@會(huì)使限制邏輯接近打印語(yǔ)句。

fib = fibonacci(100)

接下來，我將使用循環(huán)跳過前 10 個(gè)元素。

for _ in range(10):
    next(fib)

next()函數(shù)實(shí)際上是循環(huán)始終用于推進(jìn)迭代的函數(shù)。在生成器的情況下，這將返回由yield返回的任何值。在這種情況下，由于我們還不關(guān)心這些值，我們只是將它們?nèi)拥簦▽?duì)它們什么都不做）。

順便說一句，我也可以這樣調(diào)用fib.__next__()——但我更喜歡采取的更簡(jiǎn)潔方法next(fib)。它通常取決于個(gè)人偏好。兩者同樣有效。

我現(xiàn)在準(zhǔn)備好從生成器訪問一些值，但不是全部。因此，我仍將使用range()，并直接使用next()從生成器中檢索值。

for n in range(10, 21):
    print(f"{n}th value: {next(fib)}")

這可以很好地打印出所需的值：

10th value: 89
11th value: 144
12th value: 233
13th value: 377
14th value: 610
15th value: 987
16th value: 1597
17th value: 2584
18th value: 4181
19th value: 6765
20th value: 10946

還記得我們之前將限制設(shè)置為 100，現(xiàn)在已經(jīng)完成了我們的生成器，但我們不應(yīng)該直接離開并讓它等待另一個(gè)next()調(diào)用！我們程序的其余部分處于空閑狀態(tài)就會(huì)浪費(fèi)資源（盡管很少）。

相反，我們可以手動(dòng)告訴我們的生成器我們已經(jīng)完成了它。

fib.close()

這將手動(dòng)關(guān)閉生成器，就像它已經(jīng)到達(dá)一個(gè)return語(yǔ)句一樣。它現(xiàn)在可以由垃圾收集器清理。

認(rèn)識(shí)協(xié)程

生成器允許我們快速定義一個(gè)在調(diào)用之間存儲(chǔ)其狀態(tài)的可迭代對(duì)象。但是，如果我們想要相反的結(jié)果：傳遞信息并讓函數(shù)耐心等待它得到它呢？Python為此提供了協(xié)程。

對(duì)于已經(jīng)有點(diǎn)熟悉協(xié)程的人，你應(yīng)該明白我所指的是簡(jiǎn)單的協(xié)程（盡管我只是為了讀者的理智而自始至終都在說“協(xié)程”。）如果你已經(jīng)看過任何使用并發(fā)的 Python 代碼，你可能已經(jīng)遇到過它的小弟，原生協(xié)程（也稱為“異步協(xié)程”）。

現(xiàn)在，了解簡(jiǎn)單協(xié)程和原生協(xié)程都被官方認(rèn)為是“協(xié)程”，它們有很多共同的原則；原生協(xié)程建立在簡(jiǎn)單協(xié)程引入的概念之上。我們會(huì)在后續(xù)的文章中討論async。

同樣，現(xiàn)在假設(shè)當(dāng)我說“協(xié)程”時(shí)，我指的是一個(gè)簡(jiǎn)單的協(xié)程。

想象一下，你想找到一堆字符串之間的所有共同字母，比如一本書籍中那些有趣的人物名字。你不知道有多少字符串，它們會(huì)在運(yùn)行時(shí)輸入，不一定是一次全部輸入。

顯然，這種方法必須：

• 可重復(fù)使用。

• 有狀態(tài)（到目前為止共有的字母。）

• 本質(zhì)上是迭代的，因?yàn)槲覀儾恢牢覀儠?huì)得到多少個(gè)字符串。

普通的函數(shù)并不適合這種情況，因?yàn)槲覀儽仨氁淮螌⑺袛?shù)據(jù)作為列表或元組傳遞，而且它們本身不存儲(chǔ)狀態(tài)。同時(shí)，生成器不能處理輸入，除非是第一次調(diào)用。

我們可以嘗試新建一個(gè)類，盡管有很多模板。不管怎樣，讓我們從這兒開始，只是為了更好地掌握我們正在處理的內(nèi)容。

在我的第一個(gè)版本中，我將對(duì)傳遞給類的列表進(jìn)行修改，因此我可以隨時(shí)查看結(jié)果。如果我堅(jiān)持使用類實(shí)現(xiàn)，我可能不會(huì)那樣做，但它是實(shí)現(xiàn)我們目的最小的可行類了。此外，它在功能上與我們稍后將要編寫的協(xié)程相同，這用來比較實(shí)現(xiàn)方法很有用。

class CommonLetterCounter:

    def __init__(self, results):
        self.letters = {}
        self.counted = []
        self.results = results
        self.i = 0

    def add_word(self, word):
        word = word.lower()
        for c in word:
            if c.isalpha():
                if c not in self.letters:
                    self.letters[c] = 0
                self.letters[c] += 1

        self.counted = sorted(self.letters.items(), key=lambda kv: kv[1])
        self.counted = self.counted[::-1]

        self.results.clear()
        for item in self.counted:
            self.results.append(item)


names = ['Skimpole', 'Sloppy', 'Wopsle', 'Toodle', 'Squeers',
         'Honeythunder', 'Tulkinghorn', 'Bumble', 'Wegg',
         'Swiveller', 'Sweedlepipe', 'Jellyby', 'Smike', 'Heep',
         'Sowerberry', 'Pumblechook', 'Podsnap', 'Tox', 'Wackles',
         'Scrooge', 'Snodgrass', 'Winkle', 'Pickwick']

results = []
counter = CommonLetterCounter(results)

for name in names:
    counter.add_word(name)

for letter, count in results:
    print(f'{letter} apppears {count} times.')

根據(jù)我的輸出，這本數(shù)據(jù)特別喜歡帶有 e、o、s、l 和 p 的名字。誰(shuí)知道？

我們可以使用協(xié)程完成相同的結(jié)果。

def count_common_letters(results):
    letters = {}

    while True:
        word = yield
        word = word.lower()
        for c in word:
            if c.isalpha():
                if c not in letters:
                    letters[c] = 0
                letters[c] += 1

        counted = sorted(letters.items(), key=lambda kv: kv[1])
        counted = counted[::-1]

        results.clear()
        for item in counted:
            results.append(item)


names = ['Skimpole', 'Sloppy', 'Wopsle', 'Toodle', 'Squeers',
         'Honeythunder', 'Tulkinghorn', 'Bumble', 'Wegg',
         'Swiveller', 'Sweedlepipe', 'Jellyby', 'Smike', 'Heep',
         'Sowerberry', 'Pumblechook', 'Podsnap', 'Tox', 'Wackles',
         'Scrooge', 'Snodgrass', 'Winkle', 'Pickwick']

results = []
counter = count_common_letters(results)
counter.send(None)  # prime the coroutine

for name in names:
    counter.send(name)  # send data to the coroutine

counter.close()  # manually end the coroutine

for letter, count in results:
    print(f'{letter} apppears {count} times.')

讓我們仔細(xì)看看這里發(fā)生了什么。乍一看，協(xié)程與函數(shù)并沒有什么不同，但與生成器一樣，yield關(guān)鍵字的使用就大不相同了。

在協(xié)程中，yield它代表“等到你的輸入，然后在這里使用它”。

你會(huì)注意到兩種方法之間的大多數(shù)處理邏輯是相同的。我們只是取消了類模板。我們存儲(chǔ)協(xié)程的實(shí)例就像存儲(chǔ)對(duì)象一樣，只是為了確保每次向它發(fā)送更多數(shù)據(jù)時(shí)都使用相同的實(shí)例。

類和協(xié)程之間的主要區(qū)別在于用法。我們使用協(xié)程的send()函數(shù)向協(xié)程發(fā)送數(shù)據(jù)：

for name in names:
    counter.send(name)

在我們這樣做之前，我們必須首先調(diào)用（上面使用counter.send(None)的）或counter.__next__()。協(xié)程不能立即接收值；它必須首先運(yùn)行它的所有代碼，直到它的第一個(gè)yield.

與生成器一樣，協(xié)程在到達(dá)其正常執(zhí)行流程的末尾或到達(dá)return語(yǔ)句時(shí)完成。由于在我們的示例中這些情況都沒有發(fā)生的機(jī)會(huì)，所以我選擇手動(dòng)關(guān)閉協(xié)程：

counter.close()

簡(jiǎn)而言之，使用協(xié)程：

• 將其實(shí)例保存為變量，例如counter，

• 用counter.send(None),counter.__next__()或next(counter)輸入?yún)f(xié)程，

• 用counter.send()發(fā)送數(shù)據(jù)，

• 如有必要，用counter.close()關(guān)閉它。

嘗試協(xié)程

還記得關(guān)于生成器的規(guī)則，不能將 yield放在語(yǔ)句的try子句中try-finally嗎？但是這里不適用！因?yàn)?code>yield在協(xié)程中的行為非常不同（處理傳入數(shù)據(jù)，而不是傳出數(shù)據(jù)），以這種方式使用它是完全可以接受的。

throw()

生成器和協(xié)程也有一個(gè)throw()函數(shù)，用于在它們暫停的地方引發(fā)異常。你會(huì)從《錯(cuò)誤和異常》一文中了解到，異?？梢杂米鞔a執(zhí)行流程的正常部分。

例如，假設(shè)你想將數(shù)據(jù)發(fā)送到遠(yuǎn)程服務(wù)器。你現(xiàn)在已經(jīng)有一個(gè)連接對(duì)象，并且已使用協(xié)程通過該連接發(fā)送數(shù)據(jù)。

在你的代碼中，當(dāng)檢測(cè)到你已經(jīng)失去了網(wǎng)絡(luò)連接，但是由于你與服務(wù)器的通信方式，協(xié)程發(fā)送的所有數(shù)據(jù)都會(huì)毫無(wú)保留的被丟棄。

考慮一下下面這個(gè)我已經(jīng)刪除的示例代碼。（假設(shè)實(shí)際的連接邏輯本身不適合處理回退或報(bào)告連接錯(cuò)誤。）

class Connection:
    """ Stub object simulating connection to a server """

    def __init__(self, addr):
        self.addr = addr

    def transmit(self, data):
        print(f"X: {data[0]}, Y: {data[1]} sent to {self.addr}")


def send_to_server(conn):
    """ Coroutine demonstrating sending data """
    while True:
        raw_data = yield
        raw_data = raw_data.split(' ')
        coords = (float(raw_data[0]), float(raw_data[1]))
        conn.transmit(coords)


conn = Connection("example.com")

sender = send_to_server(conn)
sender.send(None)

for i in range(1, 6):
    sender.send(f"{100/i} {200/i}")

# Simulate connection error...
conn.addr = None
# ...but assume the sender knows nothing about it.

for i in range(1, 6):
    sender.send(f"{100/i} {200/i}")

運(yùn)行該示例，我們看到前五個(gè)send()調(diào)用轉(zhuǎn)到example.com，但后五個(gè)調(diào)用轉(zhuǎn)到None。這顯然是不行的——我們想拋出問題，然后開始將數(shù)據(jù)寫到文件中，這樣它就不會(huì)永遠(yuǎn)丟失。

這就是throw()的作用。一旦我們知道我們已經(jīng)失去了連接，我們就可以提醒協(xié)程這個(gè)事實(shí)，讓它做出適當(dāng)?shù)捻憫?yīng)。

我們首先在協(xié)程中添加一個(gè)try-except：

def send_to_server(conn):
    while True:
        try:
            raw_data = yield
            raw_data = raw_data.split(' ')
            coords = (float(raw_data[0]), float(raw_data[1]))
            conn.transmit(coords)
        except ConnectionError:
            print("Oops! Connection lost. Creating fallback.")
            # Create a fallback connection!
            conn = Connection("local file")

我們的使用示例只需要進(jìn)行一處更改：一旦我們知道我們失去了連接，我們就使用sender.throw(ConnectionError)拋出異常：

conn = Connection("example.com")

sender = send_to_server(conn)
sender.send(None)

for i in range(1, 6):
    sender.send(f"{100/i} {200/i}")

# Simulate connection error...
conn.addr = None
# ...but assume the sender knows nothing about it.

sender.throw(ConnectionError) # ALERT THE SENDER!

for i in range(1, 6):
    sender.send(f"{100/i} {200/i}")

這樣的話！現(xiàn)在我們會(huì)在協(xié)程收到警報(bào)后立即收到有關(guān)連接問題的消息，并將相關(guān)錯(cuò)誤內(nèi)容寫入到本地文件，也就是所謂的日志文件。

yield from

使用生成器或協(xié)程時(shí)，你不僅限于yield，你還可以使用yield from.

例如，假設(shè)我想重寫我的斐波那契數(shù)列以使其沒有限制，并且我只想編碼前五個(gè)值。

def fibonacci():
    starter = [1, 1, 2, 3, 5]
    yield from starter

    n1 = starter[-2]
    n2 = starter[-1]

    while True:
        yield (n := n1 + n2)
        n1, n2 = n2, n

在這種情況下，yield from暫時(shí)移交給另一個(gè)可迭代對(duì)象，無(wú)論它是容器、對(duì)象還是另一個(gè)生成器。一旦該可迭代對(duì)象結(jié)束，該生成器就會(huì)啟動(dòng)并像往常一樣繼續(xù)運(yùn)行。

僅僅使用這個(gè)生成器，你不會(huì)知道它在部分時(shí)間內(nèi)使用了另一個(gè)迭代器。它只是像往常一樣工作。

fib = fibonacci()

for n in range(1,11):
    print(f"{n}th value: {next(fib)}")

fib.close()

協(xié)程也可以以類似的方式進(jìn)行切換。例如，在我們的 連接示例中，如果我們創(chuàng)建第二個(gè)協(xié)程來處理將數(shù)據(jù)寫入文件會(huì)怎樣？如果我們遇到連接錯(cuò)誤，我們可以切換到在幕后使用它。

class Connection:
    """ Stub object simulating connection to a server """

    def __init__(self, addr):
        self.addr = addr

    def transmit(self, data):
        print(f"X: {data[0]}, Y: {data[1]} sent to {self.addr}")


def save_to_file():
    while True:
        raw_data = yield
        raw_data = raw_data.split(' ')
        coords = (float(raw_data[0]), float(raw_data[1]))
        print(f"X: {coords[0]}, Y: {coords[1]} sent to local file")


def send_to_server(conn):
    while True:
        if conn is None:
            yield from save_to_file()
        else:
            try:
                raw_data = yield
                raw_data = raw_data.split(' ')
                coords = (float(raw_data[0]), float(raw_data[1]))
                conn.transmit(coords)
            except ConnectionError:
                print("Oops! Connection lost. Using fallback.")
                conn = None


conn = Connection("example.com")

sender = send_to_server(conn)
sender.send(None)

for i in range(1, 6):
    sender.send(f"{100/i} {200/i}")

# Simulate connection error...
conn.addr = None
# ...but assume the sender knows nothing about it.

sender.throw(ConnectionError) # ALERT THE SENDER!

for i in range(1, 6):
    sender.send(f"{100/i} {200/i}")

生成器和協(xié)程結(jié)合使用

你可能想知道：“我可以像從生成器中那樣直接從協(xié)程中組合兩個(gè)返回?cái)?shù)據(jù)嗎？”

我在寫這篇文章時(shí)也對(duì)此感到好奇，顯然你可以。這一切都與識(shí)別函數(shù)何時(shí)被視為生成器而不是協(xié)程有關(guān)。

關(guān)鍵很簡(jiǎn)單：實(shí)際上__next__()。send(None)在協(xié)程中同樣有效。

def count_common_letters():
    letters = {}

    word = yield
    while word is not None:
        word = word.lower()
        for c in word:
            if c.isalpha():
                if c not in letters:
                    letters[c] = 0
                letters[c] += 1
        word = yield

    counted = sorted(letters.items(), key=lambda kv: kv[1])
    counted = counted[::-1]

    for item in counted:
        yield item


names = ['Skimpole', 'Sloppy', 'Wopsle', 'Toodle', 'Squeers',
         'Honeythunder', 'Tulkinghorn', 'Bumble', 'Wegg',
         'Swiveller', 'Sweedlepipe', 'Jellyby', 'Smike', 'Heep',
         'Sowerberry', 'Pumblechook', 'Podsnap', 'Tox', 'Wackles',
         'Scrooge', 'Snodgrass', 'Winkle', 'Pickwick']

counter = count_common_letters()
counter.send(None)

for name in names:
    counter.send(name)

for letter, count in counter:
    print(f'{letter} apppears {count} times.')

我只需要觀察協(xié)程何時(shí)開始接收None（當(dāng)然是在初始啟動(dòng)之后）。由于我在word中存儲(chǔ)了yield的結(jié)果，因此我可以用word變成None時(shí)作為跳出循環(huán)的判斷條件。

當(dāng)我們將協(xié)程轉(zhuǎn)化為生成器時(shí)，它需要在yield開始輸出數(shù)據(jù)之前處理單個(gè)send(None) 。在調(diào)用我們的協(xié)程時(shí)，我們?cè)谇袚Q使用之前從未明確地send(None)；Python 在后臺(tái)執(zhí)行此操作。

另外，請(qǐng)記住協(xié)程/生成器仍然是一個(gè)函數(shù)。它只是在每次遇到yield時(shí)暫停。在我的示例中，我不能突然回去使用counter作為協(xié)程，因?yàn)闆]有執(zhí)行流程可以讓我回到word = yield。其實(shí)完全可以實(shí)現(xiàn)它，以便你可以來回切換，但如果它以犧牲可讀性或變得過于復(fù)雜為代價(jià)，則可能不明智。

到此，相信大家對(duì)“Python生成器和協(xié)程怎么用”有了更深的了解，不妨來實(shí)際操作一番吧！這里是億速云網(wǎng)站，更多相關(guān)內(nèi)容可以進(jìn)入相關(guān)頻道進(jìn)行查詢，關(guān)注我們，繼續(xù)學(xué)習(xí)！