# kafka

**Repository Path**: wudilele/kafka

## Basic Information

- **Project Name**: kafka
- **Description**: No description available
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2023-11-23
- **Last Updated**: 2023-11-24

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# kafka
Based on [php-rdkafka](https://github.com/arnaud-lb/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());
```



* 消费者
```
```