oemsaas / kafka
Based on rdkafka, kafka advanced consumers, low-level consumers and producers
v1.1.1
2023-11-24 03:51 UTC
Requires
- php: >=7.3
- ext-rdkafka: *
README
Based on php-rdkafka, kafka advanced consumers, low-level consumers and producers
installation
composer require oemsaas/kafka
example
生产者
/* * SASL生产者 */ $conf = Conf::newSimpleSASLKafkaConf($brokers, $username, $password); $conf->setAckMode(1); $conf->set('retries', '1'); $conf->set('retry.backoff.ms', '100'); $conf->set('socket.timeout.ms', '1000'); $conf->setSendCallback(function (\RdKafka\Producer $kafka, \RdKafka\Message $message) { throw new KafkaException("消息发送失败【{$message->errstr()}】"); }); $conf->setErrorCallback(function (\RdKafka\Producer $kafka, int $err, string $reason) { $message = sprintf("Kafka error: %s (reason: %s)\n", rd_kafka_err2str($err), $reason); throw new KafkaException("消息发送严重错误【{$message}】"); }); return new Producer($conf->getKafkaConfInstance());
/* * 普通生产者 */ $conf = Conf::newSimpleKafkaConf($brokers); $conf->setAckMode(1); $conf->set('retries', '1'); $conf->set('retry.backoff.ms', '100'); $conf->set('socket.timeout.ms', '1000'); $conf->setSendCallback(function (\RdKafka\Producer $kafka, \RdKafka\Message $message) { throw new KafkaException("消息发送失败【{$message->errstr()}】"); }); $conf->setErrorCallback(function (\RdKafka\Producer $kafka, int $err, string $reason) { $message = sprintf("Kafka error: %s (reason: %s)\n", rd_kafka_err2str($err), $reason); throw new KafkaException("消息发送严重错误【{$message}】"); }); return new Producer($conf->getKafkaConfInstance());
- 消费者