Bài 11: Async Programming Basics - Lập Trình Bất Đồng Bộ

Mục Tiêu Bài Học

Sau khi hoàn thành bài này, bạn sẽ:

✅ Hiểu synchronous vs asynchronous
✅ Sử dụng async/await keywords
✅ Làm việc với asyncio basics
✅ Tạo async functions
✅ Run async code
✅ Áp dụng vào I/O operations

Synchronous vs Asynchronous

Synchronous (Đồng Bộ)

import time # Synchronous - blockingdef download_file(name):    print(f"Starting download: {name}")    time.sleep(2)  # Simulate download    print(f"Finished download: {name}") # Downloads happen one by onestart = time.time()download_file("file1.txt")download_file("file2.txt")download_file("file3.txt")end = time.time() print(f"Total time: {end - start:.2f}s")  # ~6 seconds# Starting download: file1.txt# Finished download: file1.txt# Starting download: file2.txt# Finished download: file2.txt# Starting download: file3.txt# Finished download: file3.txt# Total time: 6.00s

Asynchronous (Bất Đồng Bộ)

import asyncio # Asynchronous - non-blockingasync def download_file(name):    print(f"Starting download: {name}")    await asyncio.sleep(2)  # Simulate download (non-blocking)    print(f"Finished download: {name}") # Downloads happen concurrentlyasync def main():    start = time.time()        # Run tasks concurrently    await asyncio.gather(        download_file("file1.txt"),        download_file("file2.txt"),        download_file("file3.txt")    )        end = time.time()    print(f"Total time: {end - start:.2f}s")  # ~2 seconds asyncio.run(main())# Starting download: file1.txt# Starting download: file2.txt# Starting download: file3.txt# Finished download: file1.txt# Finished download: file2.txt# Finished download: file3.txt# Total time: 2.00s

Key Difference: Async allows other tasks to run while waiting for I/O!

async/await Keywords

Creating Async Functions

import asyncio # Regular functiondef regular_function():    return "Hello" # Async function (coroutine)async def async_function():    return "Hello Async" # Call regular functionresult = regular_function()print(result)  # Hello # Async function returns coroutine objectcoro = async_function()print(coro)  # <coroutine object async_function at 0x...> # Need to await or run itresult = asyncio.run(async_function())print(result)  # Hello Async

await - Waiting for Coroutines

import asyncio async def fetch_data():    print("Fetching data...")    await asyncio.sleep(1)  # Simulate I/O    return {"user": "Alice", "age": 25} async def process_data():    # await pauses execution until coroutine completes    data = await fetch_data()    print(f"Processing: {data}")    return data # Run async functionasyncio.run(process_data())# Fetching data...# Processing: {'user': 'Alice', 'age': 25}

Multiple await Calls

import asyncio async def task1():    print("Task 1 start")    await asyncio.sleep(1)    print("Task 1 done")    return "Result 1" async def task2():    print("Task 2 start")    await asyncio.sleep(2)    print("Task 2 done")    return "Result 2" async def main():    # Sequential - waits for each    result1 = await task1()    result2 = await task2()    print(f"Results: {result1}, {result2}")    # Total time: ~3 seconds asyncio.run(main())# Task 1 start# Task 1 done# Task 2 start# Task 2 done# Results: Result 1, Result 2

asyncio Basics

asyncio.run() - Run Async Function

import asyncio async def hello():    print("Hello")    await asyncio.sleep(1)    print("World") # Run async functionasyncio.run(hello())# Hello# (wait 1 second)# World

asyncio.gather() - Run Multiple Coroutines

import asyncioimport time async def task(name, delay):    print(f"{name} start")    await asyncio.sleep(delay)    print(f"{name} done")    return f"{name} result" async def main():    start = time.time()        # Run concurrently with gather    results = await asyncio.gather(        task("Task1", 2),        task("Task2", 1),        task("Task3", 3)    )        end = time.time()    print(f"Results: {results}")    print(f"Total time: {end - start:.2f}s")  # ~3 seconds (max delay) asyncio.run(main())# Task1 start# Task2 start# Task3 start# Task2 done# Task1 done# Task3 done# Results: ['Task1 result', 'Task2 result', 'Task3 result']# Total time: 3.00s

asyncio.create_task() - Create Tasks

import asyncio async def task(name, delay):    print(f"{name} start")    await asyncio.sleep(delay)    print(f"{name} done")    return f"{name} result" async def main():    # Create tasks (starts immediately)    task1 = asyncio.create_task(task("Task1", 2))    task2 = asyncio.create_task(task("Task2", 1))        # Do other work...    print("Tasks created, doing other work...")        # Wait for tasks to complete    result1 = await task1    result2 = await task2        print(f"Results: {result1}, {result2}") asyncio.run(main())# Task1 start# Task2 start# Tasks created, doing other work...# Task2 done# Task1 done# Results: Task1 result, Task2 result

asyncio.sleep() - Non-blocking Sleep

import asyncioimport time async def count(name, n):    for i in range(n):        print(f"{name}: {i+1}")        await asyncio.sleep(0.5)  # Non-blocking async def main():    start = time.time()        # Both count concurrently    await asyncio.gather(        count("Counter1", 3),        count("Counter2", 3)    )        end = time.time()    print(f"Total: {end - start:.2f}s")  # ~1.5s (not 3s) asyncio.run(main())# Counter1: 1# Counter2: 1# Counter1: 2# Counter2: 2# Counter1: 3# Counter2: 3# Total: 1.50s

Real-world Examples

1. Concurrent API Calls

import asyncioimport aiohttp  # pip install aiohttpimport time async def fetch_url(session, url):    """Fetch URL asynchronously."""    print(f"Fetching: {url}")    async with session.get(url) as response:        data = await response.text()        print(f"Finished: {url} ({len(data)} bytes)")        return url, len(data) async def fetch_all(urls):    """Fetch multiple URLs concurrently."""    async with aiohttp.ClientSession() as session:        tasks = [fetch_url(session, url) for url in urls]        results = await asyncio.gather(*tasks)        return results # Usageurls = [    'https://www.python.org',    'https://www.github.com',    'https://www.stackoverflow.com'] start = time.time()results = asyncio.run(fetch_all(urls))end = time.time() print(f"\nTotal time: {end - start:.2f}s")print(f"Results: {results}")

2. Database Queries

import asyncio # Simulated async database operationsasync def query_users():    """Query users from database."""    print("Querying users...")    await asyncio.sleep(1)  # Simulate DB query    return [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}] async def query_posts():    """Query posts from database."""    print("Querying posts...")    await asyncio.sleep(1.5)  # Simulate DB query    return [{"id": 1, "title": "Post 1"}, {"id": 2, "title": "Post 2"}] async def query_comments():    """Query comments from database."""    print("Querying comments...")    await asyncio.sleep(0.8)  # Simulate DB query    return [{"id": 1, "text": "Comment 1"}] async def load_dashboard():    """Load dashboard data concurrently."""    print("Loading dashboard...")        # Run all queries concurrently    users, posts, comments = await asyncio.gather(        query_users(),        query_posts(),        query_comments()    )        return {        "users": users,        "posts": posts,        "comments": comments    } # Sequential would take ~3.3s, concurrent takes ~1.5sdata = asyncio.run(load_dashboard())print(f"\nDashboard data: {data}")

3. Web Scraper

import asyncioimport aiohttpfrom bs4 import BeautifulSoup class AsyncScraper:    def __init__(self, urls):        self.urls = urls        self.results = []        async def fetch_page(self, session, url):        """Fetch and parse a page."""        try:            async with session.get(url, timeout=10) as response:                html = await response.text()                return url, html        except Exception as e:            print(f"Error fetching {url}: {e}")            return url, None        async def scrape_all(self):        """Scrape all URLs concurrently."""        async with aiohttp.ClientSession() as session:            tasks = [self.fetch_page(session, url) for url in self.urls]            results = await asyncio.gather(*tasks)                        for url, html in results:                if html:                    # Parse HTML (simplified)                    print(f"Scraped {url}: {len(html)} bytes")                    self.results.append({"url": url, "size": len(html)})                return self.results # Usageurls = [    'https://example.com/page1',    'https://example.com/page2',    'https://example.com/page3'] scraper = AsyncScraper(urls)results = asyncio.run(scraper.scrape_all())

4. Task Queue Worker

import asynciofrom collections import deque class AsyncTaskQueue:    def __init__(self, num_workers=3):        self.queue = asyncio.Queue()        self.num_workers = num_workers        self.results = []        async def worker(self, name):        """Worker to process tasks."""        while True:            task = await self.queue.get()                        if task is None:  # Poison pill                break                        print(f"{name} processing: {task}")            await asyncio.sleep(1)  # Simulate work            result = f"{task} processed by {name}"            self.results.append(result)                        self.queue.task_done()        async def run(self, tasks):        """Run queue with workers."""        # Add tasks to queue        for task in tasks:            await self.queue.put(task)                # Create workers        workers = [            asyncio.create_task(self.worker(f"Worker-{i}"))            for i in range(self.num_workers)        ]                # Wait for all tasks to complete        await self.queue.join()                # Stop workers        for _ in range(self.num_workers):            await self.queue.put(None)                await asyncio.gather(*workers)                return self.results # Usagequeue = AsyncTaskQueue(num_workers=3)tasks = [f"Task-{i}" for i in range(10)]results = asyncio.run(queue.run(tasks))print(f"\nResults: {results}")

5. Rate-Limited API Client

import asyncioimport time class RateLimitedClient:    def __init__(self, requests_per_second=5):        self.delay = 1.0 / requests_per_second        self.last_request = 0        async def request(self, url):        """Make rate-limited request."""        # Wait if needed to respect rate limit        now = time.time()        time_since_last = now - self.last_request                if time_since_last < self.delay:            wait_time = self.delay - time_since_last            await asyncio.sleep(wait_time)                self.last_request = time.time()                # Make request        print(f"Requesting: {url}")        await asyncio.sleep(0.5)  # Simulate request        return f"Response from {url}"        async def fetch_all(self, urls):        """Fetch all URLs with rate limiting."""        tasks = [self.request(url) for url in urls]        results = await asyncio.gather(*tasks)        return results # Usage - max 5 requests per secondclient = RateLimitedClient(requests_per_second=5)urls = [f"https://api.example.com/endpoint{i}" for i in range(10)] start = time.time()results = asyncio.run(client.fetch_all(urls))end = time.time() print(f"\nTotal time: {end - start:.2f}s")

Error Handling

import asyncio async def failing_task():    """Task that raises exception."""    await asyncio.sleep(1)    raise ValueError("Task failed!") async def successful_task():    """Task that succeeds."""    await asyncio.sleep(1)    return "Success" async def main():    try:        # gather with return_exceptions=True        results = await asyncio.gather(            successful_task(),            failing_task(),            return_exceptions=True        )                for i, result in enumerate(results):            if isinstance(result, Exception):                print(f"Task {i} failed: {result}")            else:                print(f"Task {i} succeeded: {result}")        except Exception as e:        print(f"Error: {e}") asyncio.run(main())# Task 0 succeeded: Success# Task 1 failed: Task failed!

Async Context Managers

import asyncio class AsyncDatabase:    """Async context manager."""        async def __aenter__(self):        print("Connecting to database...")        await asyncio.sleep(0.5)        print("Connected!")        return self        async def __aexit__(self, exc_type, exc_val, exc_tb):        print("Closing database connection...")        await asyncio.sleep(0.5)        print("Closed!")        async def query(self, sql):        print(f"Executing: {sql}")        await asyncio.sleep(1)        return [{"id": 1, "name": "Alice"}] async def main():    async with AsyncDatabase() as db:        results = await db.query("SELECT * FROM users")        print(f"Results: {results}") asyncio.run(main())# Connecting to database...# Connected!# Executing: SELECT * FROM users# Results: [{'id': 1, 'name': 'Alice'}]# Closing database connection...# Closed!

When to Use Async

✅ Good Use Cases

# I/O-bound operationsasync def good_async():    # Network requests    await fetch_url("https://api.example.com")        # File I/O    await read_file_async("data.txt")        # Database queries    await db.query("SELECT * FROM users")        # Sleep/delays    await asyncio.sleep(1)

❌ Bad Use Cases

# CPU-bound operations - use multiprocessing insteadasync def bad_async():    # Heavy computation    result = sum(i**2 for i in range(10000000))  # Blocks event loop!        # Use threading/multiprocessing for CPU-bound    # Or use asyncio.to_thread() for blocking sync code

Best Practices

import asyncio # 1. Use async/await consistentlyasync def good():    result = await async_function()  # Good    return result # 2. Don't mix blocking and asyncasync def bad():    time.sleep(1)  # Bad - blocks event loop    # Use: await asyncio.sleep(1) # 3. Use asyncio.gather() for concurrent tasksasync def concurrent():    results = await asyncio.gather(        task1(),        task2(),        task3()    ) # 4. Handle exceptions properlyasync def safe():    try:        result = await risky_task()    except Exception as e:        print(f"Error: {e}") # 5. Use asyncio.create_task() for fire-and-forgetasync def fire_and_forget():    task = asyncio.create_task(background_task())    # Continue without waiting # 6. Use async context managersasync def with_resources():    async with AsyncResource() as resource:        await resource.use() # 7. Set timeoutsasync def with_timeout():    try:        result = await asyncio.wait_for(slow_task(), timeout=5.0)    except asyncio.TimeoutError:        print("Task timed out")

Bài Tập Thực Hành

Bài 1: Async Downloader

Tạo concurrent file downloader:

Download multiple files
Show progress
Handle errors
Retry failed downloads

Bài 2: API Client

Build async API client:

Multiple endpoints
Rate limiting
Error handling
Response caching

Bài 3: Web Crawler

Create async web crawler:

Crawl multiple pages
Follow links
Extract data
Respect robots.txt

Bài 4: Task Scheduler

Implement async task scheduler:

Schedule tasks
Priority queue
Concurrent execution
Result collection

Bài 5: Chat Server

Build simple async chat:

Multiple clients
Broadcast messages
Handle connections
Async I/O

Tóm Tắt

✅ Async vs Sync: Async cho I/O-bound, sync cho CPU-bound
✅ async/await: Define và call async functions
✅ asyncio.run(): Run main async function
✅ asyncio.gather(): Run multiple coroutines concurrently
✅ asyncio.create_task(): Create task immediately
✅ Use cases: API calls, database, web scraping, I/O
✅ Best practices: Don't block event loop, handle errors

Bài Tiếp Theo

Bài 12: Type Hints và Annotations - Type hints, typing module, mypy, và static type checking! 🚀

Remember:

Use async for I/O-bound operations
Don't block the event loop
Handle exceptions properly
Use asyncio.gather() for concurrency
async/await makes concurrent code readable! 🎯

#asyncio

Async Programming Basics - Lập Trình Bất Đồng Bộ Cơ Bản