* * Examples: * php worker.php email-service orders * php worker.php analytics-service orders */ require_once __DIR__ . '/../autoload.php'; $group = $argv[1] ?? 'worker-1'; $topic = $argv[2] ?? 'orders'; $config = new KafkaConfig('127.0.0.1:9092'); $consumer = new Consumer($config, $group, [$topic]); echo "Consumer group '{$group}' listening on '{$topic}'... (Ctrl+C to stop)\n"; // Stop the loop cleanly on Ctrl+C so the last message finishes + commits. pcntl_async_signals(true); pcntl_signal(SIGINT, static fn() => $consumer->stop()); $consumer->consume(function (array $msg) use ($group) { // ---- 1) take your action on the message ---- printf( "[%s] processing: topic=%s partition=%d offset=%d key=%s payload=%s\n", $group, $msg['topic'], $msg['partition'], $msg['offset'], $msg['key'] ?? '(none)', $msg['payload'] ); // ...do the real work here (DB write, API call, etc.)... // ---- 2) when this function returns without throwing, the library // commits the offset = "all done" acknowledgement to Kafka ---- }); $consumer->close(); echo "[{$group}] stopped.\n";