Asyncio: Giới thiệu

Định nghĩa

Mô-đun asyncio cung cấp một khung làm việc xoay quanh vòng lặp sự kiện (event loop). Vòng lặp sự kiện cơ bản là chờ thứ gì đó xảy ra rồi phản ứng với sự kiện đó. Nó chịu trách nhiệm xử lý các sự kiện như I/O và sự kiện hệ thống.

asyncio thực tế có sẵn một số hiện thực vòng lặp khác nhau. Mô-đun sẽ mặc định chọn hiện thực có khả năng hiệu quả nhất cho hệ điều hành mà nó đang chạy; tuy nhiên bạn có thể chọn rõ vòng lặp sự kiện nếu muốn. Một vòng lặp sự kiện về cơ bản là “khi sự kiện A xảy ra, phản ứng bằng hàm B”.

Hãy nghĩ về một máy chủ khi nó chờ ai đó truy cập và yêu cầu một tài nguyên, chẳng hạn như một trang web. Nếu trang web không quá phổ biến, máy chủ sẽ rỗi trong một thời gian dài. Nhưng khi có truy cập, máy chủ cần phải phản ứng. Sự phản ứng này được gọi là xử lý sự kiện. Khi người dùng tải trang web, máy chủ sẽ kiểm tra và gọi một hoặc nhiều bộ xử lý sự kiện. Khi các bộ xử lý sự kiện đó hoàn tất, chúng cần trả quyền điều khiển về cho vòng lặp sự kiện. Để làm điều này trong Python, asyncio sử dụng coroutine.

Coroutine là một hàm đặc biệt có thể nhường quyền điều khiển cho bên gọi mà không mất trạng thái của nó. Coroutine là một “consumer” và là phần mở rộng của generator. Một trong những lợi ích lớn của chúng so với luồng (thread) là chúng không dùng nhiều bộ nhớ để thực thi. Lưu ý rằng khi bạn gọi một hàm coroutine, nó thực ra không thực thi ngay. Thay vào đó, nó trả về một đối tượng coroutine mà bạn có thể chuyển cho vòng lặp sự kiện để chạy ngay lập tức hoặc sau đó.

Một thuật ngữ khác bạn có thể gặp khi dùng mô-đun asyncio là future. Future về cơ bản là một đối tượng đại diện cho kết quả của công việc chưa hoàn thành. Vòng lặp sự kiện của bạn có thể theo dõi các đối tượng future và chờ chúng hoàn tất. Khi một future hoàn tất, nó được đặt thành trạng thái done. Asyncio cũng hỗ trợ lock và semaphore.

Mảnh ghép cuối cùng tôi muốn nhắc đến là Task. Task là một wrapper cho coroutine và là một lớp con của Future. Bạn thậm chí có thể lập lịch cho một Task bằng vòng lặp sự kiện.

async và await

Từ khóa async và await được thêm vào Python 3.5 để định nghĩa coroutine gốc (native) và biến chúng thành một kiểu riêng biệt so với coroutine dựa trên generator. Nếu bạn muốn đọc mô tả chuyên sâu về async và await, hãy xem PEP 492.

Trong Python 3.4, bạn sẽ tạo một coroutine như sau:

# Python 3.4 coroutine example
# Import `asyncio`
import _______
 
@asyncio.coroutine
def my_coro():
    yield from func()

Decorator này vẫn hoạt động trong Python 3.5, nhưng mô-đun types đã được cập nhật dưới dạng một hàm coroutine có thể cho bạn biết thứ bạn tương tác có phải coroutine gốc hay không. Bắt đầu từ Python 3.5, bạn có thể dùng async def để định nghĩa một hàm coroutine về mặt cú pháp.

Vì vậy hàm ở trên sẽ trở thành như sau:

import _______
 
async def _______():
    await func()

Khi bạn định nghĩa một coroutine theo cách này, bạn không thể dùng yield bên trong hàm coroutine. Thay vào đó nó phải bao gồm câu lệnh return hoặc await để trả giá trị cho bên gọi. Lưu ý rằng từ khóa await chỉ có thể dùng bên trong hàm async def.

Cặp từ khóa async/await có thể được xem là một API dùng cho lập trình bất đồng bộ. Mô-đun asyncio chỉ là một khung làm việc tình cờ sử dụng async/await để lập trình bất đồng bộ. Thực tế có một dự án tên là curio chứng minh khái niệm này vì nó là một hiện thực riêng của vòng lặp sự kiện sử dụng async/await ở bên dưới.

Ví dụ Coroutine tệ

Dù việc có nhiều thông tin nền về cách mọi thứ hoạt động chắc chắn là hữu ích, đôi khi bạn chỉ muốn xem vài ví dụ để cảm nhận cú pháp và cách ghép nối mọi thứ.

Với suy nghĩ đó, hãy bắt đầu bằng một ví dụ đơn giản!

Một tác vụ khá phổ biến mà bạn sẽ muốn thực hiện là tải xuống một tệp từ đâu đó, dù là tài nguyên nội bộ hay một tệp trên Internet. Thông thường bạn sẽ muốn tải xuống nhiều hơn một tệp.

Vậy hãy tạo một cặp coroutine có thể làm điều đó:

import asyncio
import os
import urllib.request
 
async def download_coroutine(url):
    #"A coroutine to download the specified url"
    request = urllib.request.urlopen(url)
    filename = os.path.basename(url)
 
    with open(filename, 'wb') as file_handle:
        while True:
            chunk = request.read(1024)
            if not chunk:
                break
            file_handle.write(chunk)
    msg = 'Finished downloading {filename}'.format(filename=filename)
    return msg
 
async def main(urls):
    """
    Creates a group of coroutines and waits for them to finish
    """
    coroutines = [download_coroutine(url) for url in urls]
    completed, pending = await asyncio.wait(coroutines)
    for item in completed:
        print(item.result())
 
 
if __name__ == '__main__':
    urls = ["http://www.irs.gov/pub/irs-pdf/f1040.pdf",
            "http://www.irs.gov/pub/irs-pdf/f1040a.pdf",
            "http://www.irs.gov/pub/irs-pdf/f1040ez.pdf",
            "http://www.irs.gov/pub/irs-pdf/f1040es.pdf",
            "http://www.irs.gov/pub/irs-pdf/f1040sb.pdf"]
 
    event_loop = asyncio.get_event_loop()
    try:
        event_loop.run_until_complete(main(urls))
    finally:
        event_loop.close()Được hỗ trợ bởi

Trong đoạn mã này, chúng ta nhập các mô-đun cần thiết rồi tạo coroutine đầu tiên bằng cú pháp async. Coroutine này tên là download_coroutine và nó dùng urllib của Python để tải xuống bất kỳ URL nào được truyền vào. Khi xong, nó sẽ trả về một thông báo cho biết như vậy.

Coroutine còn lại là coroutine chính. Nó cơ bản nhận một danh sách một hoặc nhiều URL và xếp hàng chúng. Chúng ta dùng hàm wait của asyncio để chờ các coroutine hoàn tất. Tất nhiên, để thực sự bắt đầu các coroutine, chúng cần được thêm vào vòng lặp sự kiện. Ta làm điều đó ở cuối cùng, nơi ta lấy một vòng lặp sự kiện và gọi phương thức run_until_complete của nó. Bạn sẽ thấy rằng chúng ta truyền coroutine chính vào vòng lặp sự kiện. Điều này bắt đầu chạy coroutine chính, cái sẽ xếp hàng coroutine thứ hai và khởi động nó. Đây được gọi là coroutine xâu chuỗi (chained).

Vấn đề với ví dụ này là nó thực sự không phải coroutine. Lý do là hàm download_coroutine không bất đồng bộ. Vấn đề ở đây là urllib không bất đồng bộ và hơn nữa, tôi không dùng await hay yield ở đâu cả. Cách tốt hơn là dùng gói aiohttp. Cùng xem tiếp!

Ví dụ Coroutine tốt hơn

Gói aiohttp được thiết kế để tạo các máy khách và máy chủ HTTP bất đồng bộ. Bạn có thể cài đặt bằng pip như sau:

pip install aiohttpĐược hỗ trợ bởi

Sau khi cài xong, hãy cập nhật mã của chúng ta để dùng aiohttp nhằm tải các tệp:

import aiohttp
import asyncio
import async_timeout
import os
 
async def download_coroutine(session, url):
    with async_timeout.timeout(10):
        async with session.get(url) as response:
            filename = __.path.basename(url)
            with open(________, 'wb') as f_handle:
                while True:
                    chunk = await response.content.read(1024)
                    if not chunk:
                        break
                    f_handle.write(_____)
            return await response.release()
 
async def main(loop):
    urls = ["http://www.irs.gov/pub/irs-pdf/f1040.pdf",
        "http://www.irs.gov/pub/irs-pdf/f1040a.pdf",
        "http://www.irs.gov/pub/irs-pdf/f1040ez.pdf",
        "http://www.irs.gov/pub/irs-pdf/f1040es.pdf",
        "http://www.irs.gov/pub/irs-pdf/f1040sb.pdf"]
 
    async with aiohttp.ClientSession(loop=loop) as session:
        tasks = [download_coroutine(session, url) for url in urls]
        await asyncio.gather(*tasks)
 
if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main(loop))

Bạn sẽ thấy ở đây chúng ta nhập thêm vài thành phần mới: aiohttp và async_timeout. Thành phần sau thực ra là một phụ thuộc của aiohttp và cho phép chúng ta tạo một context manager timeout. Hãy bắt đầu từ cuối mã và đi ngược lên. Ở điều kiện cuối, chúng ta khởi động vòng lặp sự kiện bất đồng bộ và gọi hàm chính.

Trong hàm chính, chúng ta tạo một đối tượng ClientSession rồi truyền nó cho hàm coroutine tải xuống đối với mỗi URL muốn tải. Trong download_coroutine, chúng ta tạo một context manager async_timeout.timeout() cơ bản tạo một bộ đếm thời gian X giây. Khi hết thời gian, context manager kết thúc hoặc hết hạn. Ở đây, timeout là 10 giây. Tiếp theo chúng ta gọi phương thức get() của session để nhận được một đối tượng response. Giờ đến phần có chút “ma thuật”. Khi bạn dùng thuộc tính content của đối tượng response, nó trả về một thể hiện của aiohttp.StreamReader cho phép chúng ta tải tệp theo từng khối với kích thước tùy ý. Khi đọc tệp, ta ghi nó ra đĩa cục bộ. Cuối cùng ta gọi phương thức release() của response, cái sẽ hoàn tất việc xử lý phản hồi.

Theo tài liệu của aiohttp, bởi vì đối tượng response được tạo trong một context manager, về mặt kỹ thuật nó sẽ gọi release() một cách ngầm định. Nhưng trong Python, tường minh thường tốt hơn, và có một ghi chú trong tài liệu rằng chúng ta không nên dựa vào việc kết nối tự mất đi, nên tôi cho rằng trong trường hợp này tốt hơn là tự gọi release.

Vẫn còn một phần bị chặn ở đây, đó là đoạn mã thực sự ghi ra đĩa. Trong khi đang ghi tệp, chúng ta vẫn bị chặn. Có một thư viện khác tên là aiofiles mà chúng ta có thể dùng để cố gắng khiến việc ghi tệp cũng bất đồng bộ, nhưng tôi sẽ để bản cập nhật đó cho bạn đọc.

Lập lịch lời gọi

Bạn cũng có thể lập lịch gọi tới các hàm thông thường bằng vòng lặp sự kiện asyncio. Phương thức đầu tiên chúng ta xem là call_soon. Phương thức call_soon về cơ bản sẽ gọi callback hoặc bộ xử lý sự kiện của bạn sớm nhất có thể. Nó hoạt động như một hàng đợi FIFO, vì vậy nếu một số callback mất thời gian chạy, thì các callback khác sẽ bị trì hoãn cho đến khi những cái trước hoàn tất.

Hãy xem một ví dụ:

import ______
import functools
 
def event_handler(loop, stop=False):
    print('Event handler called')
    if stop:
        print('stopping the loop')
        loop.stop()

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    try:
        loop.call_soon(functools.partial(_____________, ____))
        print('starting event loop')
        loop.call_soon(functools.partial(_____________, ____, stop=True))
 
        loop.__________()
    finally:
        print('closing event loop')
        loop._____()

Phần lớn các hàm của asyncio không chấp nhận đối số từ khóa, nên chúng ta sẽ cần mô-đun functools nếu cần truyền từ khóa cho bộ xử lý sự kiện. Hàm thông thường của chúng ta sẽ in ra stdout một số văn bản khi được gọi. Nếu bạn đặt đối số stop của nó thành True, nó cũng sẽ dừng vòng lặp sự kiện.

Lần gọi đầu tiên, chúng ta không dừng vòng lặp. Lần gọi thứ hai, chúng ta dừng vòng lặp. Lý do muốn dừng vòng lặp là vì chúng ta đã yêu cầu nó run_forever, thứ sẽ đặt vòng lặp sự kiện vào vòng lặp vô hạn. Khi vòng lặp dừng, chúng ta có thể đóng nó.

Nếu bạn chạy đoạn mã này, bạn sẽ thấy kết quả sau:

starting event loop
Event handler called
Event handler called
stopping the loop
closing event loopĐược hỗ trợ bởi

Có một hàm liên quan tên là call_soon_threadsafe. Đúng như tên gọi, nó hoạt động giống call_soon nhưng an toàn cho luồng. Nếu bạn muốn thực sự trì hoãn một lời gọi đến một thời điểm trong tương lai, bạn có thể dùng hàm call_later. Trong trường hợp này, chúng ta có thể đổi chữ ký call_soon thành như sau:

loop.call_later(1, event_handler, loop)Được hỗ trợ bởi

Điều này sẽ trì hoãn việc gọi bộ xử lý sự kiện của chúng ta một giây, sau đó nó sẽ gọi và truyền vòng lặp vào làm tham số đầu tiên. Nếu bạn muốn lập lịch vào một thời điểm cụ thể trong tương lai, thì bạn cần lấy thời gian của vòng lặp thay vì thời gian của máy tính. Bạn có thể làm như sau:

current_time = loop.time()Được hỗ trợ bởi

Khi đã có, bạn chỉ cần dùng hàm call_at và truyền vào thời điểm muốn gọi bộ xử lý sự kiện. Giả sử chúng ta muốn gọi bộ xử lý sự kiện sau năm phút nữa. Bạn có thể làm như sau:

loop.call_at(current_time + 300, event_handler, loop)Được hỗ trợ bởi

Trong ví dụ này, chúng ta dùng thời gian hiện tại vừa lấy và cộng thêm 300 giây, tức năm phút. Bằng cách đó, chúng ta trì hoãn việc gọi bộ xử lý sự kiện trong năm phút! Khá thú vị!

Task

Task là một lớp con của Future và là một wrapper quanh coroutine. Chúng cho bạn khả năng theo dõi khi nào chúng hoàn tất xử lý. Vì chúng là một kiểu Future, các coroutine khác có thể chờ một task, và bạn cũng có thể lấy kết quả của một task khi nó đã xử lý xong.

Hãy xem một ví dụ đơn giản:

import _______

async def my_task(seconds):
    """
    A task to do for a number of seconds
    """
    print('This task is taking {} seconds to complete'.format(
        seconds))
    await asyncio.sleep(_______)
    return 'task finished'
 
 
if __name__ == '__main__':
    my_event_loop = asyncio.get_event_loop()
    try:
        print('task creation started')
        task_obj = my_event_loop.create_task(my_task(seconds=2))
        my_event_loop.run_until_complete(task_obj)
    finally:
        my_event_loop._____()
 
    print("The task's result was: {}".format(task_obj.result()))

Ở đây chúng ta tạo một hàm bất đồng bộ nhận số giây mà hàm sẽ cần để chạy. Điều này mô phỏng một tiến trình chạy lâu. Sau đó chúng ta tạo vòng lặp sự kiện và tạo một đối tượng task bằng cách gọi hàm create_task của đối tượng vòng lặp sự kiện. Hàm create_task nhận hàm mà chúng ta muốn chuyển thành task. Rồi chúng ta yêu cầu vòng lặp sự kiện chạy cho đến khi task hoàn tất. Cuối cùng, chúng ta lấy kết quả của task vì nó đã xong.

Các task cũng có thể bị hủy rất dễ bằng cách dùng phương thức cancel của chúng. Chỉ cần gọi nó khi bạn muốn kết thúc một task. Nếu một task bị hủy khi đang chờ một thao tác khác, task sẽ phát sinh CancelledError.

Tổng kết

Đến đây, bạn đã biết đủ để bắt đầu tự làm việc với thư viện asyncio. Thư viện asyncio rất mạnh mẽ và cho phép bạn thực hiện nhiều tác vụ thú vị. Thư viện asyncio được thiết kế cho socket mạng.

Một thư viện tuyệt vời cho phép lập trình socket bất đồng bộ trước khi có asyncio là framework Twisted. Một dự án thú vị khác là Dask, một thư viện tính toán song song linh hoạt cho tính toán phân tích. Hãy đào sâu tài liệu của mô-đun asyncio và để ý tưởng của bạn tuôn chảy!