Kafka>> php-kafka-lib>> 返回
项目作者： jobcloud

项目描述 ：
PHP Kafka producer / consumer library with PHP Avro support, based on php-rdkafka
高级语言： PHP
项目地址: git://github.com/jobcloud/php-kafka-lib.git
创建时间： 2019-12-08T10:23:41Z
项目社区：https://github.com/jobcloud/php-kafka-lib
开源协议：MIT License
下载


php-kafka-lib

CircleCI
Maintainability
Test Coverage
Latest Stable Version
Latest Unstable Version

Description

This is a library that makes it easier to use Kafka in your PHP project.

This library relies on arnaud-lb/php-rdkafka
Avro support relies on flix-tech/avro-serde-php
The documentation of the php extension,
can help out to understand the internals of this library.

Requirements

  • php: ^8.0
  • ext-rdkafka: >=4.0.0
  • librdkafka: >=0.11.6 (if you use <librdkafka:1.x please define your own error callback)

:warning: To use the transactional producer you’ll need:

  • ext-rdkafka: >=4.1.0
  • librdkafka: >=1.4

Installation

  1. composer require jobcloud/php-kafka-lib "~1.0"

Enable Avro support

If you need Avro support, run:

  1. composer require flix-tech/avro-serde-php "~1.4"

Usage

Producer

Kafka

Simple example
  1. <?php
  3. use Jobcloud\Kafka\Message\KafkaProducerMessage;
  4. use Jobcloud\Kafka\Producer\KafkaProducerBuilder;
  6. $producer = KafkaProducerBuilder::create()
  7.     ->withAdditionalBroker('localhost:9092')
  8.     ->build();
  10. $message = KafkaProducerMessage::create('test-topic', 0)
  11.             ->withKey('asdf-asdf-asfd-asdf')
  12.             ->withBody('some test message payload')
  13.             ->withHeaders([ 'key' => 'value' ]);
  15. $producer->produce($message);
  17. // Shutdown producer, flush messages that are in queue. Give up after 20s
  18. $result = $producer->flush(20000);
Transactional producer (needs >=php-rdkafka:4.1 and >=librdkafka:1.4)
  1. <?php
  3. use Jobcloud\Kafka\Message\KafkaProducerMessage;
  4. use Jobcloud\Kafka\Producer\KafkaProducerBuilder;
  5. use Jobcloud\Kafka\Exception\KafkaProducerTransactionRetryException;
  6. use Jobcloud\Kafka\Exception\KafkaProducerTransactionAbortException;
  7. use Jobcloud\Kafka\Exception\KafkaProducerTransactionFatalException;
  9. $producer = KafkaProducerBuilder::create()
  10.     ->withAdditionalBroker('localhost:9092')
  11.     ->build();
  13. $message = KafkaProducerMessage::create('test-topic', 0)
  14.             ->withKey('asdf-asdf-asfd-asdf')
  15.             ->withBody('some test message payload')
  16.             ->withHeaders([ 'key' => 'value' ]);
  17. try {
  18.     $producer->beginTransaction(10000);
  19.     $producer->produce($message);
  20.     $producer->commitTransaction(10000);
  21. } catch (KafkaProducerTransactionRetryException $e) {
  22.     // something went wrong but you can retry the failed call (either beginTransaction or commitTransaction)
  23. } catch (KafkaProducerTransactionAbortException $e) {
  24.     // you need to call $producer->abortTransaction(10000); and try again
  25. } catch (KafkaProducerTransactionFatalException $e) {
  26.     // something went very wrong, re-create your producer, otherwise you could jeopardize the idempotency guarantees
  27. }
  29. // Shutdown producer, flush messages that are in queue. Give up after 20s
  30. $result = $producer->flush(20000);
Avro Producer

To create an avro prodcuer add the avro encoder. 

  1. <?php
  3. use FlixTech\AvroSerializer\Objects\RecordSerializer;
  4. use Jobcloud\Kafka\Message\KafkaProducerMessage;
  5. use Jobcloud\Kafka\Message\Encoder\AvroEncoder;
  6. use Jobcloud\Kafka\Message\Registry\AvroSchemaRegistry;
  7. use Jobcloud\Kafka\Producer\KafkaProducerBuilder;
  8. use Jobcloud\Kafka\Message\KafkaAvroSchema;
  9. use FlixTech\SchemaRegistryApi\Registry\CachedRegistry;
  10. use FlixTech\SchemaRegistryApi\Registry\BlockingRegistry;
  11. use FlixTech\SchemaRegistryApi\Registry\PromisingRegistry;
  12. use FlixTech\SchemaRegistryApi\Registry\Cache\AvroObjectCacheAdapter;
  13. use GuzzleHttp\Client;
  15. $cachedRegistry = new CachedRegistry(
  16.     new BlockingRegistry(
  17.         new PromisingRegistry(
  18.             new Client(['base_uri' => 'jobcloud-kafka-schema-registry:9081'])
  19.         )
  20.     ),
  21.     new AvroObjectCacheAdapter()
  22. );
  24. $registry = new AvroSchemaRegistry($cachedRegistry);
  25. $recordSerializer = new RecordSerializer($cachedRegistry);
  27. //if no version is defined, latest version will be used
  28. //if no schema definition is defined, the appropriate version will be fetched form the registry
  29. $registry->addBodySchemaMappingForTopic(
  30.     'test-topic',
  31.     new KafkaAvroSchema('bodySchemaName' /*, int $version, AvroSchema $definition */)
  32. );
  33. $registry->addKeySchemaMappingForTopic(
  34.     'test-topic',
  35.     new KafkaAvroSchema('keySchemaName' /*, int $version, AvroSchema $definition */)
  36. );
  38. // if you are only encoding key or value, you can pass that mode as additional third argument
  39. // per default both key and body will get encoded
  40. $encoder = new AvroEncoder($registry, $recordSerializer /*, AvroEncoderInterface::ENCODE_BODY */);
  42. $producer = KafkaProducerBuilder::create()
  43.     ->withAdditionalBroker('kafka:9092')
  44.     ->withEncoder($encoder)
  45.     ->build();
  47. $schemaName = 'testSchema';
  48. $version = 1;
  49. $message = KafkaProducerMessage::create('test-topic', 0)
  50.             ->withKey('asdf-asdf-asfd-asdf')
  51.             ->withBody(['name' => 'someName'])
  52.             ->withHeaders([ 'key' => 'value' ]);
  54. $producer->produce($message);
  56. // Shutdown producer, flush messages that are in queue. Give up after 20s
  57. $result = $producer->flush(20000);

NOTE: To improve producer latency you can install the pcntl extension.
The php-kafka-lib already has code in place, similarly described here:
https://github.com/arnaud-lb/php-rdkafka#performance--low-latency-settings

Consumer

Kafka High Level

  1. <?php
  3. use Jobcloud\Kafka\Consumer\KafkaConsumerBuilder;
  4. use Jobcloud\Kafka\Exception\KafkaConsumerConsumeException;
  5. use Jobcloud\Kafka\Exception\KafkaConsumerEndOfPartitionException;
  6. use Jobcloud\Kafka\Exception\KafkaConsumerTimeoutException;
  8. $consumer = KafkaConsumerBuilder::create()
  9.      ->withAdditionalConfig(
  10.         [
  11.             'compression.codec' => 'lz4',
  12.             'auto.commit.interval.ms' => 500
  13.         ]
  14.     )
  15.     ->withAdditionalBroker('kafka:9092')
  16.     ->withConsumerGroup('testGroup')
  17.     ->withAdditionalSubscription('test-topic')
  18.     ->build();
  20. $consumer->subscribe();
  22. while (true) {
  23.     try {
  24.         $message = $consumer->consume();
  25.         // your business logic
  26.         $consumer->commit($message);
  27.     } catch (KafkaConsumerTimeoutException $e) {
  28.         //no messages were read in a given time
  29.     } catch (KafkaConsumerEndOfPartitionException $e) {
  30.         //only occurs if enable.partition.eof is true (default: false)
  31.     } catch (KafkaConsumerConsumeException $e) {
  32.         // Failed
  33.     }
  34. }

Kafka Low Level

  1. <?php
  3. use Jobcloud\Kafka\Consumer\KafkaConsumerBuilder;
  4. use Jobcloud\Kafka\Exception\KafkaConsumerConsumeException;
  5. use Jobcloud\Kafka\Exception\KafkaConsumerEndOfPartitionException;
  6. use Jobcloud\Kafka\Exception\KafkaConsumerTimeoutException;
  8. $consumer = KafkaConsumerBuilder::create()
  9.      ->withAdditionalConfig(
  10.         [
  11.             'compression.codec' => 'lz4',
  12.             'auto.commit.interval.ms' => 500
  13.         ]
  14.     )
  15.     ->withAdditionalBroker('kafka:9092')
  16.     ->withConsumerGroup('testGroup')
  17.     ->withAdditionalSubscription('test-topic')
  18.     ->withConsumerType(KafkaConsumerBuilder::CONSUMER_TYPE_LOW_LEVEL)
  19.     ->build();
  21. $consumer->subscribe();
  23. while (true) {
  24.     try {
  25.         $message = $consumer->consume();
  26.         // your business logic
  27.         $consumer->commit($message);
  28.     } catch (KafkaConsumerTimeoutException $e) {
  29.         //no messages were read in a given time
  30.     } catch (KafkaConsumerEndOfPartitionException $e) {
  31.         //only occurs if enable.partition.eof is true (default: false)
  32.     } catch (KafkaConsumerConsumeException $e) {
  33.         // Failed
  34.     } 
  35. }

Avro Consumer

To create an avro consumer add the avro decoder. 

  1. <?php
  3. use FlixTech\AvroSerializer\Objects\RecordSerializer;
  4. use \Jobcloud\Messaging\Kafka\Consumer\KafkaConsumerBuilder;
  5. use Jobcloud\Kafka\Exception\KafkaConsumerConsumeException;
  6. use Jobcloud\Kafka\Exception\KafkaConsumerEndOfPartitionException;
  7. use Jobcloud\Kafka\Exception\KafkaConsumerTimeoutException;
  8. use Jobcloud\Kafka\Message\Decoder\AvroDecoder;
  9. use Jobcloud\Kafka\Message\KafkaAvroSchema;
  10. use Jobcloud\Kafka\Message\Registry\AvroSchemaRegistry;
  11. use FlixTech\SchemaRegistryApi\Registry\CachedRegistry;
  12. use FlixTech\SchemaRegistryApi\Registry\BlockingRegistry;
  13. use FlixTech\SchemaRegistryApi\Registry\PromisingRegistry;
  14. use FlixTech\SchemaRegistryApi\Registry\Cache\AvroObjectCacheAdapter;
  15. use GuzzleHttp\Client;
  17. $cachedRegistry = new CachedRegistry(
  18.     new BlockingRegistry(
  19.         new PromisingRegistry(
  20.             new Client(['base_uri' => 'jobcloud-kafka-schema-registry:9081'])
  21.         )
  22.     ),
  23.     new AvroObjectCacheAdapter()
  24. );
  26. $registry = new AvroSchemaRegistry($cachedRegistry);
  27. $recordSerializer = new RecordSerializer($cachedRegistry);
  29. //if no version is defined, latest version will be used
  30. //if no schema definition is defined, the appropriate version will be fetched form the registry
  31. $registry->addBodySchemaMappingForTopic(
  32.     'test-topic',
  33.     new KafkaAvroSchema('bodySchema' , 9 /* , AvroSchema $definition */)
  34. );
  35. $registry->addKeySchemaMappingForTopic(
  36.     'test-topic',
  37.     new KafkaAvroSchema('keySchema' , 9 /* , AvroSchema $definition */)
  38. );
  40. // If you are only encoding / decoding key or value, only register the schema(s) you need.
  41. // It is advised against doing that though, some tools might not play
  42. // nice if you don't fully encode your message
  43. $decoder = new AvroDecoder($registry, $recordSerializer);
  45. $consumer = KafkaConsumerBuilder::create()
  46.      ->withAdditionalConfig(
  47.         [
  48.             'compression.codec' => 'lz4',
  49.             'auto.commit.interval.ms' => 500
  50.         ]
  51.     )
  52.     ->withDecoder($decoder)
  53.     ->withAdditionalBroker('kafka:9092')
  54.     ->withConsumerGroup('testGroup')
  55.     ->withAdditionalSubscription('test-topic')
  56.     ->build();
  58. $consumer->subscribe();
  60. while (true) {
  61.     try {
  62.         $message = $consumer->consume();
  63.         // your business logic
  64.         $consumer->commit($message);
  65.     } catch (KafkaConsumerTimeoutException $e) {
  66.         //no messages were read in a given time
  67.     } catch (KafkaConsumerEndOfPartitionException $e) {
  68.         //only occurs if enable.partition.eof is true (default: false)
  69.     } catch (KafkaConsumerConsumeException $e) {
  70.         // Failed
  71.     } 
  72. }

Additional information

Replaces messaging-lib
Check Migration.md for help to migrate.