正文

轻松上手协程：10个实用编程案例详解

/0 浏览量
0827

协程（Coroutine）是一种比线程更轻量级的并发执行单元，它可以在单个线程中实现多任务处理，从而提高程序的响应性和效率。在现代编程中，协程已成为实现异步编程的关键技术。本文将详细介绍10个实用的编程案例，帮助你轻松上手协程。

案例一：网络请求异步处理

在Web开发中，网络请求是常见的操作。使用协程可以轻松实现异步发送HTTP请求，并在请求完成后处理响应。

import asyncio
import aiohttp

async def fetch_data(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    async with aiohttp.ClientSession() as session:
        data = await fetch_data(session, 'http://example.com')
        print(data)

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

案例二：数据库操作

协程在数据库操作中也表现出色，可以实现异步查询和插入数据。

import asyncio
import aiomysql

async def fetch_data(pool):
    async with pool.acquire() as conn:
        async with conn.cursor() as cur:
            await cur.execute('SELECT * FROM table')
            result = await cur.fetchall()
            return result

async def main():
    pool = await aiomysql.create_pool(host='127.0.0.1', port=3306,
                                      user='root', password='password',
                                      db='test', loop=loop)
    data = await fetch_data(pool)
    print(data)

loop.run_until_complete(main())

案例三：文件读写

使用协程可以轻松实现文件的异步读写操作。

import asyncio

async def read_file(filename):
    async with aiofiles.open(filename, 'r') as f:
        return await f.read()

async def write_file(filename, content):
    async with aiofiles.open(filename, 'w') as f:
        await f.write(content)

async def main():
    content = 'Hello, World!'
    await write_file('example.txt', content)
    print(await read_file('example.txt'))

loop.run_until_complete(main())

案例四：生成器与协程结合

生成器可以与协程结合，实现更复杂的异步操作。

import asyncio

async def fetch_all_data():
    for i in range(5):
        yield i
        await asyncio.sleep(1)

async def main():
    data = [x for x in await fetch_all_data()]
    print(data)

loop.run_until_complete(main())

案例五：异步IO与同步IO结合

在处理混合IO操作时，可以使用asyncio.run_in_executor方法将耗时操作提交给线程池执行。

import asyncio
import time

async def fetch_data():
    start_time = time.time()
    await asyncio.sleep(2)
    elapsed_time = time.time() - start_time
    print(f"Elapsed time: {elapsed_time:.2f} seconds")
    return elapsed_time

async def main():
    loop.run_in_executor(None, fetch_data)()

loop.run_until_complete(main())

案例六：协程与多线程

协程可以与多线程结合，实现更高效的并发处理。

import asyncio
import threading

async def fetch_data():
    print("Fetching data...")
    await asyncio.sleep(2)
    print("Data fetched.")

async def main():
    threads = []
    for _ in range(5):
        t = threading.Thread(target=asyncio.run, args=(fetch_data,))
        t.start()
        threads.append(t)
    for t in threads:
        t.join()

loop.run_until_complete(main())

案例七：异步任务调度

使用asyncio.wait_for方法可以设置异步任务的超时时间。

import asyncio

async def fetch_data():
    print("Fetching data...")
    await asyncio.sleep(3)
    print("Data fetched.")

async def main():
    try:
        await asyncio.wait_for(fetch_data(), timeout=2)
    except asyncio.TimeoutError:
        print("Timeout occurred!")

loop.run_until_complete(main())

案例八：异步编程与Web框架

在Web框架中，可以使用协程处理异步请求。

from fastapi import FastAPI, Request
from starlette.responses import JSONResponse

app = FastAPI()

@app.get('/')
async def read_root(request: Request):
    return JSONResponse(content={"Hello": "World!"})

loop.run_until_complete(app.run())

案例九：协程与锁

在多线程环境下，使用协程可以简化锁的使用。

import asyncio

lock = asyncio.Lock()

async def task():
    async with lock:
        print("Task acquired the lock.")
        await asyncio.sleep(1)
        print("Task released the lock.")

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.gather(task(), task(), task()))

案例十：协程与微服务

在微服务架构中，使用协程可以实现异步通信和调用。

import asyncio
import requests

async def fetch_service_data(url):
    response = await requests.get(url)
    return response.json()

async def main():
    data = await fetch_service_data('http://service1.com/api/data')
    print(data)

loop.run_until_complete(main())

通过以上10个案例，相信你已经对协程有了更深入的了解。在实际编程中，灵活运用协程可以大大提高程序的并发性能和响应速度。祝你在协程的世界里畅游无阻！

-- 展开阅读全文 --