Concurrency and scaling

What it is

A Worker runs concurrency consume loops in one isolate. Each loop reserves a message, awaits its handler, acks, and repeats. While one handler awaits I/O, the others run. Dart does not run them in parallel on several cores.

When to use it

I/O-bound tasks (HTTP calls, database queries, email): raise concurrency (8, 16, 32) in one process.
CPU-bound tasks (image processing, PDF rendering, hashing): run more worker processes, roughly one per core, with low concurrency.
Mixed workloads: separate queues with their own worker processes, sized per workload.

When not to use it

Don’t raise concurrency for CPU-bound handlers: a handler that never yields blocks every other slot in its isolate, and heartbeats and visibility-timeout touches with them.
Don’t rely on timeLimit to free a slot from a stuck CPU loop. The worker stops waiting but the slot stays busy until the handler returns.

How it works

Queue priority. queues: ['critical', 'default', 'low'] checks queues in order on every reserve, so low only runs when the others are empty. A busy high-priority queue starves the lower ones; give important low-volume work its own queue and worker if that is a problem.

Scaling out. Every worker process pulls from the shared broker; add processes or machines to add throughput. There is no coordinator. On RabbitMQ, prefetchCount (32) caps deliveries buffered per process.

Rate limits are per worker process (RateLimit.parse('10/s') with three workers is up to 30/s overall). Over-limit messages are re-enqueued with a delay instead of holding a slot.

Pausing. A control channel pause(queue) makes every subscribed worker stop reserving from that queue until resume. Running tasks finish.

Shutdown. worker.stop() stops reserving and waits for running tasks; stop(timeout:) gives up after the timeout and leaves the rest unacked for redelivery. WorkerCli does this on SIGINT/SIGTERM with a 30 s drain timeout.

Measuring. The package’s benchmark/throughput_bench.dart floods a worker and reports enqueue rate, drain rate and latency percentiles. Useful flags: --broker=memory|redis|rabbitmq, --tasks=N, --concurrency=N, --sweep (concurrency 1 to 32), --work-us=N (simulated work), --with-results.

dart run benchmark/throughput_bench.dart --broker=redis --sweep --tasks=20000 --work-us=500

Example

import 'package:gisila_queue/gisila_queue.dart';

// I/O-heavy worker: one process, many slots.
await app.worker(queues: ['emails', 'webhooks'], concurrency: 32).run();

// CPU-heavy worker: start one of these per core (separate processes).
await app.worker(queues: ['thumbnails'], concurrency: 1).run();