Impara Task Cancellation in Asyncio | Advanced Asyncio Patterns

Understanding how to cancel running tasks is crucial when working with asyncio. Sometimes, you may need to stop a task before it completes—perhaps because its result is no longer needed, it has taken too long, or your application is shutting down. In asyncio, you can cancel a task by calling its cancel() method. When a task is cancelled, it raises an asyncio.CancelledError inside the coroutine, which you can catch and handle if necessary. This allows you to clean up resources or perform any required finalization before the task fully stops. Handling cancellation properly ensures your asynchronous applications remain robust and responsive.


              12345678910111213141516171819202122
            
import asyncio

async def long_running_task():
    try:
        print("Task started.")
        await asyncio.sleep(5)
        print("Task finished.")
    except asyncio.CancelledError:
        print("Task was cancelled.")
        raise

async def main():
    task = asyncio.create_task(long_running_task())
    await asyncio.sleep(1)  # Let the task run for a bit
    print("Cancelling the task...")
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Caught task cancellation in main.")

await main()

In this example, you create a long-running task and then cancel it after one second. When the task is cancelled, it raises asyncio.CancelledError inside the coroutine, which is caught and handled. This pattern is essential for managing resources and ensuring your program responds properly to changes or user requests.

Tutto è chiaro?

Grazie per i tuoi commenti!

Sezione 3. Capitolo 1

Chieda ad AI

Chieda pure quello che desidera o provi una delle domande suggerite per iniziare la nostra conversazione

Scorri per mostrare il menu


              12345678910111213141516171819202122
            
import asyncio

async def long_running_task():
    try:
        print("Task started.")
        await asyncio.sleep(5)
        print("Task finished.")
    except asyncio.CancelledError:
        print("Task was cancelled.")
        raise

async def main():
    task = asyncio.create_task(long_running_task())
    await asyncio.sleep(1)  # Let the task run for a bit
    print("Cancelling the task...")
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Caught task cancellation in main.")

await main()

Tutto è chiaro?

Grazie per i tuoi commenti!

Sezione 3. Capitolo 1