Python異步編程之a(chǎn)syncio.create_task()用法示例解析

更新時間：2025年11月07日 11:07:03 作者：吐個泡泡v

asyncio.create_task()是Python中用于將協(xié)程包裝為任務并加入事件循環(huán)的核心方法,這篇文章主要介紹了Python異步編程之a(chǎn)syncio.create_task()用法的相關(guān)資料,文中通過代碼介紹的非常詳細,需要的朋友可以參考下

前言

asyncio.create_task() 是Python 3.7+引入的函數(shù)，用于創(chuàng)建一個Task對象。Task是對協(xié)程的封裝，可以讓協(xié)程在事件循環(huán)中并發(fā)執(zhí)行。

1. 基本語法

task = asyncio.create_task(coro)

coro: 要包裝的協(xié)程對象
返回值: Task對象

2. 基礎(chǔ)用法示例

2.1 簡單的任務創(chuàng)建

import asyncio
import time

async def say_hello(name, delay):
    print(f"開始執(zhí)行 {name}")
    await asyncio.sleep(delay)
    print(f"Hello, {name}!")
    return f"完成 {name}"

async def main():
    # 創(chuàng)建任務
    task1 = asyncio.create_task(say_hello("Alice", 2))
    task2 = asyncio.create_task(say_hello("Bob", 1))
    
    # 等待任務完成
    result1 = await task1
    result2 = await task2
    
    print(result1)
    print(result2)

# 運行主函數(shù)
asyncio.run(main())

2.2 多個任務并發(fā)執(zhí)行

import asyncio

async def fetch_data(url, delay):
    print(f"開始獲取 {url}")
    await asyncio.sleep(delay)
    return f"數(shù)據(jù)來自 {url}"

async def main():
    # 創(chuàng)建多個任務
    tasks = [
        asyncio.create_task(fetch_data("https://api1.com", 1)),
        asyncio.create_task(fetch_data("https://api2.com", 2)),
        asyncio.create_task(fetch_data("https://api3.com", 1.5))
    ]
    
    # 等待所有任務完成
    results = await asyncio.gather(*tasks)
    
    for result in results:
        print(result)

asyncio.run(main())

3. 高級用法

3.1 任務的取消

import asyncio

async def long_running_task():
    try:
        for i in range(10):
            print(f"任務執(zhí)行中... {i}")
            await asyncio.sleep(1)
        return "任務完成"
    except asyncio.CancelledError:
        print("任務被取消")
        raise

async def main():
    task = asyncio.create_task(long_running_task())
    
    # 3秒后取消任務
    await asyncio.sleep(3)
    task.cancel()
    
    try:
        await task
    except asyncio.CancelledError:
        print("捕獲到取消異常")

asyncio.run(main())

3.2 任務狀態(tài)檢查

import asyncio

async def sample_task():
    await asyncio.sleep(2)
    return "完成"

async def main():
    task = asyncio.create_task(sample_task())
    
    print(f"任務創(chuàng)建后 - 完成: {task.done()}")
    
    await asyncio.sleep(1)
    print(f"等待1秒后 - 完成: {task.done()}")
    
    await task
    print(f"任務完成后 - 完成: {task.done()}")
    print(f"任務結(jié)果: {task.result()}")

asyncio.run(main())

3.3 異常處理

import asyncio

async def task_with_exception():
    await asyncio.sleep(1)
    raise ValueError("這是一個錯誤")

async def normal_task():
    await asyncio.sleep(2)
    return "正常完成"

async def main():
    task1 = asyncio.create_task(task_with_exception())
    task2 = asyncio.create_task(normal_task())
    
    try:
        # 使用 gather 并設(shè)置 return_exceptions=True 來處理異常
        results = await asyncio.gather(task1, task2, return_exceptions=True)
        
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                print(f"任務 {i} 發(fā)生異常: {result}")
            else:
                print(f"任務 {i} 結(jié)果: {result}")
    except Exception as e:
        print(f"捕獲異常: {e}")

asyncio.run(main())

4. 實際應用場景

4.1 并發(fā)HTTP請求

示例一：

import asyncio
import aiohttp

async def fetch_url(session, url):
    try:
        async with session.get(url) as response:
            return await response.text()
    except Exception as e:
        return f"錯誤: {e}"

async def fetch_multiple_urls():
    urls = [
        'https://httpbin.org/delay/1',
        'https://httpbin.org/delay/2',
        'https://httpbin.org/delay/1'
    ]
    
    async with aiohttp.ClientSession() as session:
        # 創(chuàng)建任務列表
        tasks = [asyncio.create_task(fetch_url(session, url)) for url in urls]
        
        # 并發(fā)執(zhí)行所有任務
        results = await asyncio.gather(*tasks)
        
        for i, result in enumerate(results):
            print(f"URL {i+1}: {len(result) if isinstance(result, str) else result} 字符")

# 注意：需要安裝 aiohttp: pip install aiohttp
# asyncio.run(fetch_multiple_urls())

示例二：

import asyncio
import aiohttp


async def process_http_request():
    # 使用aiohttp進行異步HTTP請求
    timeout = aiohttp.ClientTimeout(total=1000)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        headers = {}
        data = {}
        url = ""
        is_error = False
        try:
            print(f"HTTP異步請求開始，data: {data} ")
            async with session.post(
                    url=url,
                    headers=headers,
                    json=data,
                    timeout=timeout  # 為單個請求設(shè)置超時
            ) as response:
                if response.status == 200:
                    # 獲取響應數(shù)據(jù)
                    response_data = await response.json()
                    if response_data['data']['status'] != 200:
                        print(f"HTTP異步請求處理失敗: {response_data} ")
                        is_error = True
                else:
                    print(f"HTTP異步請求處理失敗，狀態(tài)碼: {response.status}")
                    is_error = True
        except asyncio.TimeoutError:
            print(f"HTTP異步請求超時!")
            is_error = True

    print(f"HTTP異步請求完成")
    return is_error

4.2 數(shù)據(jù)庫操作并發(fā)

import asyncio
# import asyncpg  # 示例使用 asyncpg

async def query_database(query_id):
    # 模擬數(shù)據(jù)庫查詢
    await asyncio.sleep(1)  # 模擬查詢時間
    return f"查詢結(jié)果 {query_id}"

async def concurrent_queries():
    # 創(chuàng)建多個查詢?nèi)蝿?
    tasks = [
        asyncio.create_task(query_database(i)) 
        for i in range(5)
    ]
    
    # 并發(fā)執(zhí)行查詢
    results = await asyncio.gather(*tasks)
    
    for result in results:
        print(result)

asyncio.run(concurrent_queries())

create_task() vs ensure_future()

在Python 3.7之前，我們使用 asyncio.ensure_future() 來創(chuàng)建任務：

# 舊方式（仍然可用）
task = asyncio.ensure_future(coro)

# 新方式（推薦）
task = asyncio.create_task(coro)

兩者的主要區(qū)別：

create_task() 更明確地表示創(chuàng)建一個任務
ensure_future() 功能更廣泛，可以處理多種類型的awaitable對象
在現(xiàn)代Python中，推薦使用 create_task()

5. 最佳實踐

5.1 合理使用任務

import asyncio

async def good_practice():
    # 好的做法：明確創(chuàng)建任務
    task1 = asyncio.create_task(some_coroutine())
    task2 = asyncio.create_task(another_coroutine())
    
    # 等待結(jié)果
    result1 = await task1
    result2 = await task2
    
    return result1, result2

async def avoid_this():
    # 避免這樣做：直接await協(xié)程不會并發(fā)執(zhí)行
    result1 = await some_coroutine()
    result2 = await another_coroutine()
    return result1, result2

5.2 使用超時控制

import asyncio

async def slow_operation():
    await asyncio.sleep(5)
    return "完成"

async def main():
    try:
        task = asyncio.create_task(slow_operation())
        # 設(shè)置3秒超時
        result = await asyncio.wait_for(task, timeout=3.0)
        print(result)
    except asyncio.TimeoutError:
        print("操作超時")
        # 取消任務
        task.cancel()

asyncio.run(main())

6.常見問題

在Python的asyncio中，asyncio.create_task()創(chuàng)建的任務不需要被await也會運行。

asyncio任務執(zhí)行機制

當你使用asyncio.create_task()時，任務會被立即調(diào)度執(zhí)行，而不需要顯式地await它。這是因為：

asyncio.create_task()會將任務添加到事件循環(huán)中進行調(diào)度

事件循環(huán)會自動運行這些任務，不需要顯式await

await的作用是等待任務完成并獲取結(jié)果，而不是觸發(fā)任務執(zhí)行

1. 任務創(chuàng)建與調(diào)度

當調(diào)用asyncio.create_task()時，會發(fā)生以下事情：

（1）任務被調(diào)度：任務被添加到事件循環(huán)(event loop)中進行調(diào)度

（2）立即執(zhí)行：任務會在事件循環(huán)的下一個周期開始執(zhí)行

（3）并發(fā)運行：任務與當前代碼以及其他任務并發(fā)運行

2. await的作用

await關(guān)鍵字的作用是：

（1）等待任務完成：暫停當前協(xié)程直到任務完成

（2）獲取結(jié)果：獲取任務的返回值

（3）異常處理：傳播任務中發(fā)生的異常

但是await不是觸發(fā)任務執(zhí)行的必要條件。

3. 實際示例

import asyncio

async def background_task(name):
    print(f"任務 {name} 開始")
    await asyncio.sleep(2)
    print(f"任務 {name} 完成")
    return f"結(jié)果 {name}"

async def main():
    # 創(chuàng)建任務但不await - 任務會自動運行
    task1 = asyncio.create_task(background_task("1"))
    task2 = asyncio.create_task(background_task("2"))
    
    print("任務已創(chuàng)建")
    
    # 做一些其他工作
    await asyncio.sleep(1)
    print("主協(xié)程繼續(xù)執(zhí)行")
    
    # 等待任務完成獲取結(jié)果
    result1 = await task1
    result2 = await task2
    
    print(f"結(jié)果: {result1}, {result2}")

# 運行主函數(shù)
asyncio.run(main())

從輸出結(jié)果可以看出，即使沒有立即await任務，它們也已經(jīng)開始運行了。

7. 總結(jié)

asyncio.create_task() 是Python異步編程中的核心函數(shù)，學會使用能夠：

并發(fā)執(zhí)行協(xié)程：通過創(chuàng)建Task對象實現(xiàn)真正的并發(fā)
靈活控制任務：可以取消、檢查狀態(tài)、處理異常
提高程序性能：特別適合I/O密集型操作
簡化異步代碼：讓異步編程更加直觀和易用

到此這篇關(guān)于Python異步編程之a(chǎn)syncio.create_task()用法的文章就介紹到這了,更多相關(guān)Python asyncio.create_task()用法內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: