* * Examples: * php worker.php email-service orders * php worker.php analytics-service orders * * Run several at once (see notes at bottom): * - SAME group -> Kafka splits the work between them. * - DIFFERENT groups -> each gets its own copy of every message. */ require_once __DIR__ . '/../autoload.php'; $group = isset($argv[1]) ? $argv[1] : 'worker-1'; $topic = isset($argv[2]) ? $argv[2] : 'orders'; $config = new KafkaConfig('127.0.0.1:9092'); $consumer = new Consumer($config, $group, array($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. if (function_exists('pcntl_async_signals')) { pcntl_async_signals(true); pcntl_signal(SIGINT, function () use ($consumer) { $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'], isset($msg['key']) && $msg['key'] !== null ? $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";