Есть ли очередность в асинхронности?

Очередность в асинхронности

Да, в асинхронном программировании очередность существует и играет критически важную роль. Это одна из самых часто неправильно понимаемых концепций.

Что такое очередность (ordering) в асинхронности

Очередность — это гарантия, что операции выполняются в определённом порядке, несмотря на асинхронный характер их выполнения. В Python очередность управляется Event Loop и контролируется разработчиком через различные механизмы.

Event Loop и порядок выполнения

Питоновский asyncio использует event loop, который управляет порядком выполнения coroutine'ов. Если несколько coroutine'ов готовы к выполнению одновременно, loop выполняет их поочередно (по принципу FIFO очереди):

import asyncio

async def task1():
    print("Task 1 начало")
    await asyncio.sleep(0.1)
    print("Task 1 конец")

async def task2():
    print("Task 2 начало")
    await asyncio.sleep(0.05)
    print("Task 2 конец")

async def main():
    # Параллельное выполнение
    await asyncio.gather(task1(), task2())

asyncio.run(main())

# Вывод:
# Task 1 начало
# Task 2 начало
# Task 2 конец (завершилась раньше)
# Task 1 конец

Проблемы с очередностью

Основная проблема — отсутствие гарантии порядка выполнения, если используется asyncio.gather() или create_task(). Разработчик должен явно контролировать очередность.

Проблема: Race Condition

import asyncio

counter = 0

async def increment():
    global counter
    temp = counter
    await asyncio.sleep(0)  # Точка переключения
    counter = temp + 1  # Race condition!

async def main():
    global counter
    counter = 0
    # Запускаем 10 инкрементов параллельно
    await asyncio.gather(*[increment() for _ in range(10)])
    print(counter)  # Выведет < 10!

asyncio.run(main())

Вместо ожидаемого 10, выведет меньшее число, потому что несколько coroutine'ов одновременно работают с одной переменной.

Способы контроля очередности

1. Использование await для последовательного выполнения

import asyncio

async def task1():
    print("Task 1")
    await asyncio.sleep(0.1)

async def task2():
    print("Task 2")
    await asyncio.sleep(0.1)

async def main():
    # Строгая очередность: task1 → task2
    await task1()
    await task2()

asyncio.run(main())

# Вывод:
# Task 1
# Task 2

2. Использование Lock для синхронизации

import asyncio

lock = asyncio.Lock()
counter = 0

async def increment():
    global counter
    async with lock:  # Критическая секция
        temp = counter
        await asyncio.sleep(0)  # Безопасно — lock удерживается
        counter = temp + 1

async def main():
    global counter
    counter = 0
    await asyncio.gather(*[increment() for _ in range(10)])
    print(counter)  # Теперь выведет 10

asyncio.run(main())

3. Использование Semaphore для ограничения параллелизма

import asyncio

sem = asyncio.Semaphore(2)  # Максимум 2 одновременно

async def worker(n):
    async with sem:
        print(f"Worker {n} работает")
        await asyncio.sleep(1)
        print(f"Worker {n} завершен")

async def main():
    await asyncio.gather(*[worker(i) for i in range(5)])

asyncio.run(main())

# Выполнится: 2 worker'а параллельно, потом 2 следующих, потом 1

4. Использование Queue для упорядоченной обработки

import asyncio

async def producer(queue):
    for i in range(5):
        await queue.put(i)
        print(f"Отправлено: {i}")

async def consumer(queue):
    while True:
        item = await queue.get()
        print(f"Обработано: {item}")
        await asyncio.sleep(0.1)
        queue.task_done()

async def main():
    queue = asyncio.Queue()
    
    # Producer и consumer выполняются параллельно,
    # но очередность обработки гарантирована
    await asyncio.gather(
        producer(queue),
        consumer(queue),
        return_exceptions=True
    )

asyncio.run(main())

Очередность в базах данных (Database race conditions)

Особенно важна очередность при работе с БД. Используется SELECT FOR UPDATE SKIP LOCKED:

import asyncio
from sqlalchemy import select, text
from sqlalchemy.ext.asyncio import AsyncSession

async def find_and_lock_resource(session: AsyncSession, resource_id: int):
    # Блокируем строку для гарантии очередности
    result = await session.execute(
        select(Resource).where(Resource.id == resource_id).with_for_update()
    )
    resource = result.scalars().first()
    return resource

# Правильно для Telegram bot'ов
async def find_opponent(user_id: int) -> int | None:
    query = """
    SELECT user_id FROM waiting_users 
    WHERE user_id != @user_id
    ORDER BY created_at ASC
    LIMIT 1
    FOR UPDATE SKIP LOCKED
    """
    # Гарантирует, что один и тот же opponent не будет найден двумя пользователями

Выводы

• Очередность в асинхронности существует, но её нужно контролировать
• await гарантирует последовательное выполнение
• Lock, Semaphore, Queue — инструменты для синхронизации
• Race conditions могут возникать неожиданно — используй SELECT FOR UPDATE в БД
• Event Loop выполняет готовые coroutine'ы в FIFO порядке
• Помни: асинхронность ≠ параллелизм, это просто другой способ управления временем выполнения