Зачем нужен Async I/O

В production Flink джобы регулярно делают enrichment — обогащение событий из внешнего источника. Транзакция приходит с user_id, и нужно получить user_profile из Redis. Order приходит с product_id, и нужно подтянуть product_metadata из PostgreSQL. Заявка на займ требует scoring через ML model в gRPC сервисе. Все эти операции — блокирующие I/O.

Наивная реализация через обычный map/flatMap приведёт к тому, что один медленный subtask остановит весь stream. В этом уроке разберёмся, почему sync-lookup убивает throughput, и как Async I/O решает эту проблему — на интуитивном уровне, без deep dive в детали API.

Проблема: sync lookup в map

Представим, что нам нужно обогатить Transaction событие user profile из Redis:

DataStream<EnrichedTransaction> enriched = transactions
    .map(tx -> {
        UserProfile profile = redisClient.get(tx.getUserId());  // <- блокирующий вызов
        return new EnrichedTransaction(tx, profile);
    });

Каждый вызов redisClient.get(...) блокирует thread исполнения subtask’а. Пока Redis отвечает (даже 1-5 ms latency), subtask не может обработать следующее событие.

Математика throughput’а subtask’а:

• Latency Redis = 5 ms.
• Один subtask = один thread = последовательное исполнение.
• Maximum throughput одного subtask = 1 / 0.005 = 200 событий/сек.

Если в потоке 100 000 событий/сек, нужен parallelism = 500. Это безумно. И мы не учли, что Redis тоже не бесконечно масштабируется.

Решение: async I/O — много lookups одновременно

Async I/O позволяет отправить много запросов параллельно, не блокируя subtask thread. Пока 100 запросов в Redis обрабатываются, subtask может принимать новые события и отправлять новые запросы.

DataStream<EnrichedTransaction> enriched = AsyncDataStream
    .unorderedWait(
        transactions,
        new RedisEnrichmentFunction(),
        1000,  // timeout ms
        TimeUnit.MILLISECONDS,
        100    // capacity — сколько запросов параллельно
    );

Что меняется:

• Subtask отправляет до capacity запросов одновременно (например, 100).
• Каждый запрос — non-blocking (использует асинхронный клиент Redis).
• Когда любой запрос завершается — соответствующее событие отправляется в downstream.
• Subtask thread свободен между отправкой запросов и обработкой ответов.

Та же математика, async:

• Latency Redis = 5 ms (одного запроса).
• Capacity = 100 — параллельные запросы.
• Throughput одного subtask = 100 / 0.005 = 20 000 событий/сек.
• Прирост: в 100 раз по сравнению с sync.

Что должен поддерживать клиент

Async I/O работает только если ваш клиент к внешней системе предоставляет async API. Если клиент только синхронный — async I/O бесполезен (это просто sync в обёртке).

Примеры:

Чек-лист: подходит ли вам async I/O

Альтернативы async I/O

Иногда async I/O — не лучший инструмент. Альтернативы:

1. Broadcast state (модуль 08): если справочник маленький (тысячи записей) — храните его в state, никаких lookups.
2. Keyed state с side input: если событие приходит с user_id, а profile тоже приходит как отдельный stream, можно сделать coGroup/join — никаких внешних lookups.
3. Cache в operator state: если можно кешировать результаты lookups на стороне subtask (с TTL), это убирает 80% нагрузки на внешнюю систему.
4. Lookup join в Flink Table API/SQL (модуль 12): декларативный синтаксис, под капотом всё равно async I/O.

Async I/O — мощный инструмент, но не panacea. Если можно избежать внешних вызовов через broadcast или state — это всегда быстрее.

Попробуй сам

Сделай бенчмарк sync vs async на 10 000 событий:

1. Подними локальный Redis: docker run -p 6379:6379 redis:7-alpine.
2. Заполни Redis 10 000 ключей: for i in $(seq 0 9999); do redis-cli set user:$i '{"name":"u'$i'"}'; done.
3. Напиши Flink job, который читает 10 000 событий из collection source, делает sync-lookup в Redis в map, и измерь время total.
4. Напиши тот же job с async I/O (Lettuce/aioredis), capacity=100. Измерь время.
5. Сравни. Должно быть как минимум в 10x быстрее, чаще — в 50x-100x.

Ключевые выводы

1. Sync lookup в обычном map блокирует subtask thread. Throughput ограничен 1 / latency_external событий/сек на subtask.
2. Async I/O позволяет отправить до capacity запросов параллельно. Throughput растёт до capacity / latency_external.
3. Прирост обычно 50x-500x, в зависимости от capacity и latency внешней системы.
4. Работает только с async клиентами (Lettuce, asyncpg, aiohttp, aio.grpc). Оборачивать sync client в CompletableFuture бесполезно.
5. Альтернативы: broadcast state для малых справочников, keyed state + co-stream для side input, lookup join в Table API.