Concorrência e paralelismo

// Executa a tarefa para cada item, com no máximo `limite` ao mesmo tempo
async function comLimite<T, R>(
  itens: T[],
  limite: number,
  tarefa: (item: T) => Promise<R>,
): Promise<R[]> {
  const resultados = new Array<R>(itens.length);
  let proximo = 0;

  async function trabalhador() {
    while (proximo < itens.length) {
      const indice = proximo++;
      resultados[indice] = await tarefa(itens[indice]);
    }
  }

  const quantidade = Math.min(limite, itens.length);
  await Promise.all(Array.from({ length: quantidade }, trabalhador));
  return resultados;
}

const ids = Array.from({ length: 20 }, (_, i) => i + 1);

const pedidos = await comLimite(ids, 5, async (id) => {
  const resposta = await fetch(`${BASE_URL}/pedidos/${id}`, {
    signal: AbortSignal.timeout(3000),
  });
  if (!resposta.ok) throw new Error(`pedido ${id}: HTTP ${resposta.status}`);
  return resposta.json();
});

import asyncio

import httpx


async def buscar_pedidos(ids: list[int], limite: int = 5) -> list[dict]:
    semaforo = asyncio.Semaphore(limite)

    async with httpx.AsyncClient(base_url=BASE_URL, timeout=3) as cliente:

        async def buscar(pedido_id: int) -> dict:
            async with semaforo:  # no máximo `limite` ao mesmo tempo
                resposta = await cliente.get(f"/pedidos/{pedido_id}")
                resposta.raise_for_status()
                return resposta.json()

        return await asyncio.gather(*(buscar(i) for i in ids))


pedidos = asyncio.run(buscar_pedidos(list(range(1, 21))))

package pedidos

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"time"

	"golang.org/x/sync/errgroup"
)

var cliente = &http.Client{Timeout: 3 * time.Second}

func BuscarPedidos(ctx context.Context, baseURL string, ids []int) ([]map[string]any, error) {
	resultados := make([]map[string]any, len(ids))

	grupo, ctx := errgroup.WithContext(ctx)
	grupo.SetLimit(5) // no máximo 5 goroutines ao mesmo tempo

	for i, id := range ids {
		grupo.Go(func() error {
			pedido, err := buscarPedido(ctx, baseURL, id)
			if err != nil {
				return err // cancela o contexto das outras
			}
			resultados[i] = pedido // cada goroutine grava só a sua posição
			return nil
		})
	}

	return resultados, grupo.Wait()
}

func buscarPedido(ctx context.Context, baseURL string, id int) (map[string]any, error) {
	pedido, _ := http.NewRequestWithContext(ctx, "GET", fmt.Sprintf("%s/pedidos/%d", baseURL, id), nil)
	resposta, err := cliente.Do(pedido)
	if err != nil {
		return nil, err
	}
	defer resposta.Body.Close()
	if resposta.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("pedido %d: HTTP %d", id, resposta.StatusCode)
	}
	var corpo map[string]any
	return corpo, json.NewDecoder(resposta.Body).Decode(&corpo)
}

import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.Semaphore;

public class BuscaDePedidos {
    private static final HttpClient CLIENTE = HttpClient.newBuilder()
        .connectTimeout(Duration.ofSeconds(3))
        .build();

    public static List<String> buscar(String baseUrl, List<Integer> ids) throws Exception {
        var limite = new Semaphore(5); // no máximo 5 ao mesmo tempo
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            List<Future<String>> futuros = new ArrayList<>();
            for (int id : ids) {
                futuros.add(executor.submit(() -> {
                    limite.acquire();
                    try {
                        return buscarPedido(baseUrl, id);
                    } finally {
                        limite.release();
                    }
                }));
            }
            List<String> pedidos = new ArrayList<>();
            for (var futuro : futuros) {
                pedidos.add(futuro.get());
            }
            return pedidos;
        }
    }

    static String buscarPedido(String baseUrl, int id) throws Exception {
        var pedido = HttpRequest.newBuilder(URI.create(baseUrl + "/pedidos/" + id))
            .timeout(Duration.ofSeconds(3))
            .build();
        var resposta = CLIENTE.send(pedido, HttpResponse.BodyHandlers.ofString());
        if (resposta.statusCode() != 200) {
            throw new IllegalStateException("pedido " + id + ": HTTP " + resposta.statusCode());
        }
        return resposta.body();
    }
}

using System.Collections.Concurrent;
using System.Net.Http.Json;

public class BuscaDePedidos(HttpClient cliente)
{
    public async Task<IReadOnlyDictionary<int, Pedido>> Buscar(IEnumerable<int> ids)
    {
        var pedidos = new ConcurrentDictionary<int, Pedido>();
        var opcoes = new ParallelOptions { MaxDegreeOfParallelism = 5 };

        await Parallel.ForEachAsync(ids, opcoes, async (id, cancelamento) =>
        {
            var pedido = await cliente.GetFromJsonAsync<Pedido>($"/pedidos/{id}", cancelamento)
                ?? throw new InvalidOperationException($"pedido {id} vazio");
            pedidos[id] = pedido;
        });

        return pedidos;
    }
}

public record Pedido(int Id, string Status);