Run workers

You’ll build a worker binary that consumes chosen queues with a chosen concurrency, stops cleanly on SIGTERM, and can be scaled by starting more copies.

Prerequisites

Tasks registered in a shared function, as in Your first task.
Redis reachable through REDIS_URL.

1. Use WorkerCli for a standard Redis worker

import 'package:gisila_queue/gisila_queue_cli.dart';
import 'package:my_app/tasks.dart';

Future<void> main(List<String> argv) async {
  final cli = WorkerCli.parse(argv);
  if (cli.help) return print(WorkerCli.usage);
  final app = cli.buildApp();
  registerTasks(app);
  await cli.run(app);
}

WorkerCli gives you --queues, --concurrency, --namespace, --loglevel, --drain-timeout and the Redis flags, and builds the app with a broker, result backend and control channel.

export REDIS_URL=redis://:secret@localhost:6379/0
dart run bin/worker.dart --namespace=shop --queues=critical,default --concurrency=8

2. Or build the worker by hand

Do this when you need RabbitMQ, an event sink, a router, or custom Worker options.

import 'dart:io';

import 'package:gisila_queue/gisila_queue.dart';
import 'package:gisila_queue/gisila_queue_cli.dart';
import 'package:my_app/tasks.dart';

Future<void> main(List<String> argv) async {
  final args = CliArgs.parse(argv);
  final redis = args.redisTarget();
  final app = GisilaQueue.redis(
    namespace: args.stringOr('namespace', 'shop'),
    host: redis.host,
    port: redis.port,
    options: redis.options,
    events: true, // feed GisilaRose
  );
  registerTasks(app);

  final worker = Worker(
    app,
    queues: args.stringList('queues') ?? ['default'],
    concurrency: args.intOr('concurrency', 4),
    heartbeatInterval: const Duration(seconds: 5),
  )..onFailure = (msg, result) => stderr.writeln('FAILED ${msg.task}: ${result.error}');

  installShutdownHooks(() => worker.stop(timeout: const Duration(seconds: 30)));
  await worker.run();
  await app.close();
}

3. Choose queues and concurrency

List queues in priority order. Earlier queues are always drained first.
Never list the dead-letter queue (dead_letter).
I/O-bound tasks: raise --concurrency. CPU-bound: run more processes with low concurrency. See Concurrency.

4. Scale out

Start the same binary several times, on one host or many. Each pulls from the shared queues. For separate workloads, run separate worker groups:

dart run bin/worker.dart --namespace=shop --queues=emails,webhooks --concurrency=32
dart run bin/worker.dart --namespace=shop --queues=thumbnails --concurrency=1

5. Shut down cleanly

On SIGINT or SIGTERM the worker stops reserving, waits up to --drain-timeout (30 s) for running tasks, then exits. Tasks still running at the deadline stay unacked and Redis redelivers them to another worker after the visibility timeout (5 minutes by default). Give your process manager a stop timeout longer than the drain timeout.

Verify it works

dart run gisila_queue:gisila_queue inspect --namespace=shop --queues=critical,default

gisila_queue inspect — redis localhost:6379, namespace "shop"
  critical: 0 ready message(s)
  default: 0 ready message(s)
  (in-flight / unacked across all queues: 0)

Enqueue a few tasks and watch the counts drop as the worker logs task ... ✓. Press Ctrl-C and confirm the worker prints shutting down… and exits.