Python生成器和協程怎么用

本篇內容主要講解“Python生成器和協程怎么用”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實用性強。下面就讓小編來帶大家學習“Python生成器和協程怎么用”吧!

我們提供的服務有：成都網站制作、做網站、微信公眾號開發、網站優化、網站認證、秀嶼ssl等。為超過千家企事業單位解決了網站和推廣的問題。提供周到的售前咨詢和貼心的售后服務，是有科學管理、有技術的秀嶼網站制作公司

認識生成器

你將如何生成任意長度的斐波那契數列？顯然，你需要跟蹤一些數據，并且需要以某種方式對其進行操作以創建下一個元素。

你的第一直覺可能是創建一個可迭代的類，這不失是一個好方法。讓我們開始，使用我們在前面幾節中已經介紹過的內容：

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)

讓我們把它變得更緊湊。

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

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

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 行長，而類為 22 行——但它同樣可讀。

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

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

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

嘗試生成器

修訂日期：2019 年 11 月 29 日

曾經規定了yield不能出現在代碼中try子句中的try-finally中。PEP 255定義了生成器語法，解釋了原因：

難點在于不能保證生成器會被恢復，因此不能保證 finally 塊會被執行；這就違背finally的目的了。

這在 PEP 342 PEP 342中進行了更改，并在 Python 2.5 中完成。

那么，為什么要討論這樣一個古老的變化呢？簡單：直到今天，我的印象是yield無法出現在try-finally中. 一些關于該主題的文章錯誤地引用了舊規則。

把生成器作為對象

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

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

首先，我將生成器保存在一個變量中，以便我可以重用它。限制對我來說并不重要，所以我會使用大的限制。使用我的循環范圍來更容易顯示內容，因為這會使限制邏輯接近打印語句。

fib = fibonacci(100)

接下來，我將使用循環跳過前 10 個元素。

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

next()函數實際上是循環始終用于推進迭代的函數。在生成器的情況下，這將返回由yield返回的任何值。在這種情況下，由于我們還不關心這些值，我們只是將它們扔掉（對它們什么都不做）。

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

我現在準備好從生成器訪問一些值，但不是全部。因此，我仍將使用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

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

相反，我們可以手動告訴我們的生成器我們已經完成了它。

fib.close()

這將手動關閉生成器，就像它已經到達一個return語句一樣。它現在可以由垃圾收集器清理。

認識協程

生成器允許我們快速定義一個在調用之間存儲其狀態的可迭代對象。但是，如果我們想要相反的結果：傳遞信息并讓函數耐心等待它得到它呢？Python為此提供了協程。

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

現在，了解簡單協程和原生協程都被官方認為是“協程”，它們有很多共同的原則；原生協程建立在簡單協程引入的概念之上。我們會在后續的文章中討論async。

同樣，現在假設當我說“協程”時，我指的是一個簡單的協程。

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

顯然，這種方法必須：

可重復使用。
有狀態（到目前為止共有的字母。）
本質上是迭代的，因為我們不知道我們會得到多少個字符串。

普通的函數并不適合這種情況，因為我們必須一次將所有數據作為列表或元組傳遞，而且它們本身不存儲狀態。同時，生成器不能處理輸入，除非是第一次調用。

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

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

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.')

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

我們可以使用協程完成相同的結果。

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.')

讓我們仔細看看這里發生了什么。乍一看，協程與函數并沒有什么不同，但與生成器一樣，yield關鍵字的使用就大不相同了。

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

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

類和協程之間的主要區別在于用法。我們使用協程的send()函數向協程發送數據：

for name in names:
    counter.send(name)

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

與生成器一樣，協程在到達其正常執行流程的末尾或到達return語句時完成。由于在我們的示例中這些情況都沒有發生的機會，所以我選擇手動關閉協程：

counter.close()

簡而言之，使用協程：

將其實例保存為變量，例如counter，
用counter.send(None),counter.__next__()或next(counter)輸入協程，
用counter.send()發送數據，
如有必要，用counter.close()關閉它。

嘗試協程

還記得關于生成器的規則，不能將 yield放在語句的try子句中try-finally嗎？但是這里不適用！因為yield在協程中的行為非常不同（處理傳入數據，而不是傳出數據），以這種方式使用它是完全可以接受的。

throw()

生成器和協程也有一個throw()函數，用于在它們暫停的地方引發異常。你會從《錯誤和異常》一文中了解到，異常可以用作代碼執行流程的正常部分。

例如，假設你想將數據發送到遠程服務器。你現在已經有一個連接對象，并且已使用協程通過該連接發送數據。

在你的代碼中，當檢測到你已經失去了網絡連接，但是由于你與服務器的通信方式，協程發送的所有數據都會毫無保留的被丟棄。

考慮一下下面這個我已經刪除的示例代碼。（假設實際的連接邏輯本身不適合處理回退或報告連接錯誤。）

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}")

運行該示例，我們看到前五個send()調用轉到example.com，但后五個調用轉到None。這顯然是不行的——我們想拋出問題，然后開始將數據寫到文件中，這樣它就不會永遠丟失。

這就是throw()的作用。一旦我們知道我們已經失去了連接，我們就可以提醒協程這個事實，讓它做出適當的響應。

我們首先在協程中添加一個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")

我們的使用示例只需要進行一處更改：一旦我們知道我們失去了連接，我們就使用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}")

這樣的話！現在我們會在協程收到警報后立即收到有關連接問題的消息，并將相關錯誤內容寫入到本地文件，也就是所謂的日志文件。

yield from

使用生成器或協程時，你不僅限于yield，你還可以使用yield from.

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

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暫時移交給另一個可迭代對象，無論它是容器、對象還是另一個生成器。一旦該可迭代對象結束，該生成器就會啟動并像往常一樣繼續運行。

僅僅使用這個生成器，你不會知道它在部分時間內使用了另一個迭代器。它只是像往常一樣工作。

fib = fibonacci()

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

fib.close()

協程也可以以類似的方式進行切換。例如，在我們的連接示例中，如果我們創建第二個協程來處理將數據寫入文件會怎樣？如果我們遇到連接錯誤，我們可以切換到在幕后使用它。

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}")

生成器和協程結合使用

你可能想知道：“我可以像從生成器中那樣直接從協程中組合兩個返回數據嗎？”

我在寫這篇文章時也對此感到好奇，顯然你可以。這一切都與識別函數何時被視為生成器而不是協程有關。

關鍵很簡單：實際上__next__()。send(None)在協程中同樣有效。

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.')

我只需要觀察協程何時開始接收None（當然是在初始啟動之后）。由于我在word中存儲了yield的結果，因此我可以用word變成None時作為跳出循環的判斷條件。

當我們將協程轉化為生成器時，它需要在yield開始輸出數據之前處理單個send(None) 。在調用我們的協程時，我們在切換使用之前從未明確地send(None)；Python 在后臺執行此操作。

另外，請記住協程/生成器仍然是一個函數。它只是在每次遇到yield時暫停。在我的示例中，我不能突然回去使用counter作為協程，因為沒有執行流程可以讓我回到word = yield。其實完全可以實現它，以便你可以來回切換，但如果它以犧牲可讀性或變得過于復雜為代價，則可能不明智。

到此，相信大家對“Python生成器和協程怎么用”有了更深的了解，不妨來實際操作一番吧！這里是創新互聯網站，更多相關內容可以進入相關頻道進行查詢，關注我們，繼續學習！

網頁題目：Python生成器和協程怎么用
鏈接URL：http://m.kartarina.com/article26/jecdjg.html

成都網站建設公司_創新互聯，為您提供小程序開發、品牌網站制作、全網營銷推廣、靜態網站、標簽優化、網站導航

聲明：本網站發布的內容（圖片、視頻和文字）以用戶投稿、用戶轉載內容為主，如果涉及侵權請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網站立場，如需處理請聯系客服。電話：028-86922220；郵箱：631063699@qq.com。內容未經允許不得轉載，或轉載時需注明來源：創新互聯

猜你還喜歡下面的內容