simpod / kafka
PHP Kafka boilerplate wrapper around RdKafka
Fund package maintenance!
simPod
Installs: 80 849
Dependents: 1
Suggesters: 0
Security: 0
Stars: 5
Watchers: 3
Forks: 0
Open Issues: 3
Requires
- php: ^8.2
- ext-pcntl: *
- ext-rdkafka: ^6
- psr/log: ^3
- thecodingmachine/safe: ^2.2
Requires (Dev)
- doctrine/coding-standard: ^12.0
- infection/infection: ^0.29.0
- phpstan/extension-installer: ^1.1
- phpstan/phpstan: 1.12.3
- phpstan/phpstan-phpunit: 1.4.0
- phpstan/phpstan-strict-rules: 1.6.0
- phpunit/phpunit: ^11.0
- thecodingmachine/phpstan-safe-rule: ^1.0
This package is auto-updated.
Last update: 2024-11-17 16:15:59 UTC
README
Installation
Add as Composer dependency:
composer require simpod/kafka
Config Constants
Some config constants are provided like ConsumerConfig
, ProducerConfig
or CommonClientConfigs
.
However, they are copied from Java API and not all are applicable to librdkafka. Consult with librdkafka documentation before use.
Clients
Consumer
KafkaConsumer
boilerplate is available with startBatch()
method (to suplement this example in librdkafka) and with start()
. They also handle
termination signals for you.
Classic Consumer
<?php declare(strict_types=1); namespace Your\AppNamespace; use RdKafka\Message; use SimPod\Kafka\Clients\Consumer\ConsumerConfig; use SimPod\Kafka\Clients\Consumer\KafkaConsumer; final class ExampleConsumer { public function run(): void { $kafkaConsumer = new KafkaConsumer($this->getConfig(), Logger::get()); $kafkaConsumer->subscribe(['topic1']); $kafkaConsumer->start( 120 * 1000, static function (Message $message) use ($kafkaConsumer) : void { // Process message here $kafkaConsumer->commit($message); // Autocommit is disabled } ); } private function getConfig(): ConsumerConfig { $config = new ConsumerConfig(); $config->set(ConsumerConfig::BOOTSTRAP_SERVERS_CONFIG, '127.0.0.1:9092'); $config->set(ConsumerConfig::ENABLE_AUTO_COMMIT_CONFIG, false); $config->set(ConsumerConfig::CLIENT_ID_CONFIG, gethostname()); $config->set(ConsumerConfig::AUTO_OFFSET_RESET_CONFIG, 'earliest'); $config->set(ConsumerConfig::GROUP_ID_CONFIG, 'consumer_group_name'); return $config; } }
Batching Consumer
<?php declare(strict_types=1); namespace Your\AppNamespace; use RdKafka\Message; use SimPod\Kafka\Clients\Consumer\ConsumerConfig; use SimPod\Kafka\Clients\Consumer\ConsumerRecords; use SimPod\Kafka\Clients\Consumer\KafkaConsumer; final class ExampleBatchConsumer { public function run(): void { $kafkaConsumer = new KafkaConsumer($this->getConfig()); $kafkaConsumer->subscribe(['topic1']); $kafkaConsumer->startBatch( 200000, 120 * 1000, static function (Message $message): void { // Process record }, static function (ConsumerRecords $consumerRecords) use ($kafkaConsumer) : void { // Process records batch $kafkaConsumer->commit($consumerRecords->getLast()); } ); } private function getConfig(): ConsumerConfig { $config = new ConsumerConfig(); $config->set(ConsumerConfig::BOOTSTRAP_SERVERS_CONFIG, '127.0.0.1:9092'); $config->set(ConsumerConfig::ENABLE_AUTO_COMMIT_CONFIG, false); $config->set(ConsumerConfig::CLIENT_ID_CONFIG, gethostname()); $config->set(ConsumerConfig::AUTO_OFFSET_RESET_CONFIG, 'earliest'); $config->set(ConsumerConfig::GROUP_ID_CONFIG, 'consumer_group_name'); return $config; } }