Async Patterns Reference

e.g. const p = new Promise((resolve, reject) => { setTimeout(() => resolve('done'), 1000); });

e.g. fetch('/api/user') .then(res => res.json()) .then(data => console.log(data));

e.g. fetchData().catch(err => console.error('Failed:', err));

e.g. fetchData() .then(process) .catch(handleError) .finally(() => hideSpinner());

e.g. const p = Promise.resolve(42);

e.g. const p = Promise.reject(new Error('fail'));

e.g. async function getUser(id) { const res = await fetch(`/api/users/${id}`); return res.json(); }

e.g. const data = await fetchData();

e.g. try { const result = await riskyOperation(); } catch (err) { console.error('Error:', err); }

e.g. for await (const chunk of readableStream) { process(chunk); }

e.g. async function* paginate(url) { let page = 1; while (true) { const res = await fetch(`${url}?page=${page++}`); const data = await res.json(); if (!data.length) break; yield data; } }

e.g. const config = await loadConfig();

e.g. const [users, posts] = await Promise.all([ fetchUsers(), fetchPosts() ]);

e.g. const results = await Promise.allSettled([p1, p2, p3]); results.forEach(r => { if (r.status === 'fulfilled') console.log(r.value); else console.error(r.reason); });

e.g. const result = await Promise.race([ fetchData(), timeout(5000) ]);

e.g. const fastest = await Promise.any([ fetchFromCDN1(), fetchFromCDN2(), fetchFromCDN3() ]);

e.g. const { promise, resolve, reject } = Promise.withResolvers(); setTimeout(() => resolve('done'), 1000);

e.g. queueMicrotask(() => console.log('microtask'));

e.g. setTimeout(() => console.log('macrotask'), 0);

e.g. process.nextTick(() => console.log('nextTick'));

e.g. const ac = new AbortController(); fetch(url, { signal: ac.signal }); setTimeout(() => ac.abort(), 5000);

e.g. async def fetch_data(url): async with aiohttp.ClientSession() as session: async with session.get(url) as resp: return await resp.json()

e.g. result = await fetch_data('https://api.example.com/data')

e.g. asyncio.run(main())

e.g. await asyncio.sleep(1.0)

e.g. loop = asyncio.get_running_loop()

e.g. async with aiohttp.ClientSession() as session: ...

e.g. async for msg in websocket: print(msg)

e.g. results = await asyncio.gather( fetch('/users'), fetch('/posts'), fetch('/comments') )

e.g. task = asyncio.create_task(background_job()) # ... do other work ... result = await task

e.g. async with asyncio.TaskGroup() as tg: t1 = tg.create_task(fetch_users()) t2 = tg.create_task(fetch_posts()) print(t1.result(), t2.result())

e.g. result = await asyncio.wait_for(slow_op(), timeout=5.0)

e.g. done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)

e.g. for coro in asyncio.as_completed(tasks): result = await coro print(result)

e.g. queue = asyncio.Queue(maxsize=100) await queue.put(item) item = await queue.get()

e.g. sem = asyncio.Semaphore(10) async with sem: await fetch(url)

e.g. go func() { result := compute() ch <- result }()

e.g. ch := make(chan int)

e.g. ch := make(chan string, 100)

e.g. ch <- 42

e.g. result := <-ch

e.g. close(ch)

e.g. for msg := range messages { fmt.Println(msg) }

e.g. v, ok := <-ch if !ok { fmt.Println("channel closed") }

e.g. func producer(out chan<- int) { ... } func consumer(in <-chan int) { ... }

e.g. select { case msg := <-ch1: handle(msg) case msg := <-ch2: handle(msg) case <-time.After(5 * time.Second): fmt.Println("timeout") }

e.g. select { case msg := <-ch: handle(msg) default: // non-blocking }

e.g. var wg sync.WaitGroup for i := 0; i < 10; i++ { wg.Add(1) go func(id int) { defer wg.Done() work(id) }(i) } wg.Wait()

e.g. var mu sync.Mutex mu.Lock() counter++ mu.Unlock()

e.g. var once sync.Once once.Do(func() { initConfig() })

e.g. ctx, cancel := context.WithCancel(context.Background()) defer cancel() go worker(ctx)

e.g. ctx, cancel := context.WithTimeout(ctx, 5*time.Second) defer cancel()

e.g. deadline := time.Now().Add(30 * time.Second) ctx, cancel := context.WithDeadline(ctx, deadline) defer cancel()

e.g. select { case <-ctx.Done(): return ctx.Err() case result := <-ch: return result }

e.g. g, ctx := errgroup.WithContext(ctx) g.Go(func() error { return fetchUsers(ctx) }) g.Go(func() error { return fetchPosts(ctx) }) if err := g.Wait(); err != nil { log.Fatal(err) }

e.g. async fn fetch_url(url: &str) -> Result<String, reqwest::Error> { reqwest::get(url).await?.text().await }

e.g. let body = reqwest::get(url).await?.text().await?;

e.g. #[tokio::main] async fn main() { let result = fetch_data().await; println!("{:?}", result); }

e.g. let handle = tokio::spawn(async { expensive_computation().await }); let result = handle.await?;

e.g. let (users, posts) = tokio::join!( fetch_users(), fetch_posts() );

e.g. tokio::select! { val = rx.recv() => println!("received {val:?}"), _ = tokio::time::sleep(Duration::from_secs(5)) => println!("timeout"), }

e.g. tokio::time::sleep(Duration::from_millis(500)).await;

e.g. match tokio::time::timeout(Duration::from_secs(5), operation()).await { Ok(result) => handle(result), Err(_) => eprintln!("timed out"), }

e.g. let data = Arc::new(tokio::sync::Mutex::new(vec![])); let mut lock = data.lock().await; lock.push(42);

e.g. let (tx, mut rx) = tokio::sync::mpsc::channel(100); tokio::spawn(async move { tx.send("hello").await.unwrap(); }); while let Some(msg) = rx.recv().await { println!("{msg}"); }

e.g. let sem = Arc::new(Semaphore::new(10)); let permit = sem.acquire().await?; // ... do work ... drop(permit);

e.g. CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> { return fetchData(); });

e.g. CompletableFuture<Void> future = CompletableFuture.runAsync(() -> { sendNotification(); });

e.g. future.thenApply(data -> parseJson(data));

e.g. future.thenCompose(user -> fetchOrders(user.getId()));

e.g. userFuture.thenCombine(ordersFuture, (user, orders) -> { return new UserWithOrders(user, orders); });

e.g. future.exceptionally(ex -> { log.error("Failed", ex); return defaultValue; });

e.g. CompletableFuture.allOf(f1, f2, f3).thenRun(() -> { // all done });

e.g. CompletableFuture.anyOf(f1, f2).thenAccept(System.out::println);

e.g. future.orTimeout(5, TimeUnit.SECONDS);

e.g. future.completeOnTimeout(fallback, 5, TimeUnit.SECONDS);

e.g. ExecutorService pool = Executors.newFixedThreadPool(10);

e.g. ExecutorService pool = Executors.newCachedThreadPool();

e.g. Future<String> future = pool.submit(() -> fetchData());

e.g. List<Future<String>> results = pool.invokeAll(tasks);

e.g. pool.shutdown(); pool.awaitTermination(60, TimeUnit.SECONDS);

e.g. Thread.ofVirtual().name("worker").start(() -> { var result = blockingIO(); process(result); });

e.g. try (var executor = Executors.newVirtualThreadPerTaskExecutor()) { executor.submit(() -> fetchUsers()); executor.submit(() -> fetchPosts()); }

e.g. try (var scope = new StructuredTaskScope.ShutdownOnFailure()) { var user = scope.fork(() -> fetchUser(id)); var orders = scope.fork(() -> fetchOrders(id)); scope.join().throwIfFailed(); return new Response(user.get(), orders.get()); }

e.g. // JS: Promise.all(urls.map(url => fetch(url))) // Go: for _, url := range urls { go fetch(url, ch) } // Python: await asyncio.gather(*[fetch(url) for url in urls])

e.g. // Go: merge channels func fanIn(channels ...<-chan int) <-chan int { out := make(chan int) var wg sync.WaitGroup for _, ch := range channels { wg.Add(1) go func(c <-chan int) { defer wg.Done() for v := range c { out <- v } }(ch) } go func() { wg.Wait(); close(out) }() return out }

e.g. // Go: jobs := make(chan Job, 100) for w := 0; w < numWorkers; w++ { go func() { for job := range jobs { process(job) } }() }

e.g. // Go: stage1 -> stage2 -> stage3 func stage(in <-chan int) <-chan int { out := make(chan int) go func() { defer close(out) for v := range in { out <- transform(v) } }() return out }

e.g. // Go: golang.org/x/time/rate limiter := rate.NewLimiter(rate.Every(time.Second/10), 10) // 10/sec, burst 10 if err := limiter.Wait(ctx); err != nil { return err }

e.g. // Python: import time class RateLimiter: def __init__(self, max_calls, period): self.calls = [] self.max_calls = max_calls self.period = period def allow(self): now = time.time() self.calls = [t for t in self.calls if t > now - self.period] if len(self.calls) < self.max_calls: self.calls.append(now) return True return False

e.g. // JS: async function limitConcurrency(tasks, limit) { const results = []; const executing = new Set(); for (const task of tasks) { const p = task().then(r => { executing.delete(p); return r; }); executing.add(p); results.push(p); if (executing.size >= limit) await Promise.race(executing); } return Promise.all(results); }

e.g. // JS: async function retry(fn, maxRetries = 3) { for (let i = 0; i <= maxRetries; i++) { try { return await fn(); } catch (err) { if (i === maxRetries) throw err; const delay = Math.min(1000 * 2 ** i, 30000); const jitter = Math.random() * delay * 0.1; await sleep(delay + jitter); } } }

e.g. // Python: async def retry_with_deadline(fn, max_attempts=5, deadline=30.0): start = time.monotonic() for attempt in range(max_attempts): try: return await asyncio.wait_for(fn(), timeout=deadline - (time.monotonic() - start)) except Exception as e: if attempt == max_attempts - 1 or time.monotonic() - start >= deadline: raise await asyncio.sleep(min(2 ** attempt, 10))

e.g. // Pseudocode: class CircuitBreaker { state = CLOSED; failures = 0; threshold = 5; lastFailure = null; cooldown = 30s; async call(fn) { if (state === OPEN) { if (now - lastFailure > cooldown) state = HALF_OPEN; else throw new CircuitOpenError(); } try { result = await fn(); if (state === HALF_OPEN) state = CLOSED; failures = 0; return result; } catch (err) { failures++; lastFailure = now; if (failures >= threshold) state = OPEN; throw err; } } }

e.g. // Java: separate pools per downstream service ExecutorService userServicePool = Executors.newFixedThreadPool(10); ExecutorService orderServicePool = Executors.newFixedThreadPool(10);

e.g. // Go: func hedgedRequest(ctx context.Context, urls []string) (Response, error) { ctx, cancel := context.WithCancel(ctx) defer cancel() ch := make(chan Response, len(urls)) for _, url := range urls { go func(u string) { if resp, err := fetch(ctx, u); err == nil { ch <- resp } }(url) } return <-ch, nil }