Importación por lotes con límite de tasa

Gana FxDart async

Requisito

Envía nueve transacciones del libro de cuentas (están en el código de abajo) a un endpoint de importación que acepta lotes de tres, una llamada cada vez: estrictamente secuencial, nunca solapadas. Después de cada lote, registra su tamaño, su importe y el total acumulado importado hasta el momento; imprime los resúmenes de lote en orden y luego demuestra que se respetó el límite de tasa mediante el contador de máximo en vuelo (debe marcar 1).

En FxDart la política entera es la cadena: chunk(3) fija el tamaño del lote, concurrent(1) fija el ritmo y scan hila el total acumulado a través de las confirmaciones (drop(1) descarta el valor inicial del scan). El propio endpoint simula la latencia con delay y suma su lote con sumBy.

Salida esperada

importing 9 txns in batches of 3, one at a time:
  batch 1: 3 txns, $172.49 — running total $172.49
  batch 2: 3 txns, $403.65 — running total $576.14
  batch 3: 3 txns, $325.34 — running total $901.48
max batches in flight: 1

Lado a lado

Dart nativo

import 'package:collection/collection.dart';

class Txn {
  final String id;
  final double amount;
  const Txn(this.id, this.amount);
}

const txns = [
  Txn('t-01', 120.00), Txn('t-02', 42.50), Txn('t-03', 9.99),
  Txn('t-04', 310.25), Txn('t-05', 75.00), Txn('t-06', 18.40),
  Txn('t-07', 220.10), Txn('t-08', 5.25), Txn('t-09', 99.99),
];

int inFlight = 0;
int maxInFlight = 0;

/// The rate-limited import endpoint: accepts one batch per call.
Future<(int, double)> importBatch(List<Txn> batch) async {
  inFlight++;
  if (inFlight > maxInFlight) maxInFlight = inFlight;
  await Future.delayed(const Duration(milliseconds: 15));
  final ack = (batch.length, batch.fold(0.0, (sum, t) => sum + t.amount));
  inFlight--;
  return ack;
}

Future<void> main() async {
  final report = <(int, int, double, double)>[];
  var running = 0.0;
  var n = 0;
  for (final batch in txns.slices(3)) {
    final (count, amount) = await importBatch(batch);
    n++;
    running += amount;
    report.add((n, count, amount, running));
  }
  print('importing ${txns.length} txns in batches of 3, one at a time:');
  for (final (n, count, amount, running) in report) {
    print('  batch $n: $count txns, \$${amount.toStringAsFixed(2)} '
        '— running total \$${running.toStringAsFixed(2)}');
  }
  print('max batches in flight: $maxInFlight');
}

FxDart

import 'package:fxdart/fxdart.dart';

class Txn {
  final String id;
  final double amount;
  const Txn(this.id, this.amount);
}

int inFlight = 0;
int maxInFlight = 0;

/// The rate-limited import endpoint: accepts one batch per call.
Future<(int, double)> importBatch(List<Txn> batch) async {
  inFlight++;
  if (inFlight > maxInFlight) maxInFlight = inFlight;
  final ack = await delay(const Duration(milliseconds: 15),
      (batch.length, fx(batch).sumBy((t) => t.amount).toDouble()));
  inFlight--;
  return ack;
}

Future<void> main() async {
  final report = await fx(txns)
      .chunk(3)
      .toAsync()
      .map(importBatch)
      .concurrent(1)
      .scan((acc, b) => (acc.$1 + 1, b.$1, b.$2, acc.$4 + b.$2),
          (0, 0, 0.0, 0.0))
      .drop(1) // drop the scan seed
      .toList();
  print('importing ${txns.length} txns in batches of 3, one at a time:');
  for (final (n, count, amount, running) in report) {
    print('  batch $n: $count txns, \$${amount.toStringAsFixed(2)} '
        '— running total \$${running.toStringAsFixed(2)}');
  }
  print('max batches in flight: $maxInFlight');
}

Por qué difieren

Seamos justos: una importación estrictamente secuencial es la única política de concurrencia que un simple bucle for maneja con elegancia, y la versión nativa se lee bien —slices de package:collection hasta cubre el troceado en lotes—. El total acumulado, en cambio, ya es estado mutable hilado a mano (running += amount junto a n++), mientras que scan lo convierte en un paso declarado. Y la sencillez del bucle es un callejón sin salida: el día que el endpoint permita dos lotes concurrentes, la versión con FxDart cambia 1 por 2, mientras que el bucle se convierte en el pool de workers de los otros ejemplos asíncronos. La cadena enuncia la política; el bucle la codifica.

Benchmark

Apple M1 Max, 32 GB de RAM · Dart 3.12.2 (compilado AOT) · 2026-08-24

Caso async: la escala principal es N = 100,000, no 1,000,000. Cada elemento cuesta una vuelta del event loop en ambos lados, así que un millón de awaits reales mediría el event loop de Dart durante minutos — no el pipeline. Los retardos son de longitud cero y se conserva el límite de concurrencia del ejemplo; lo que comparan las barras es la maquinaria del pipeline.

N = 100

Tiempo Empate

Dart nativo 156 µs

FxDart 177 µs

Memoria pico Empate

Dart nativo 16.5 MB

FxDart 16.8 MB

N = 10,000

Tiempo Gana nativo

Dart nativo 12.7 ms

FxDart 17.3 ms

Memoria pico Gana nativo

Dart nativo 24.0 MB

FxDart 27.5 MB

N = 100,000

Tiempo Gana nativo

Dart nativo 131.0 ms

FxDart 157.7 ms

Memoria pico Empate

Dart nativo 75.9 MB

FxDart 79.7 MB

Las barras son medianas de iteraciones cronometradas repetidas en procesos nuevos por lado (los N pequeños se agrupan por resolución del temporizador). Dos lados a menos del 5% entre sí — o a menos de 0.6 ms, una diferencia que nadie puede percibir — cuentan como empate; las carreras relativas ajustadas se vuelven a medir hasta 5 veces. En una app, cualquier cosa por debajo de unos pocos milisegundos es invisible para el usuario, gane la barra que gane. La memoria es el RSS pico del proceso. La VM de Dart y el dataset son idénticos en ambos lados, así que la diferencia entre las dos barras es lo que retiene el pipeline en sí.