README

基于php-rdkafka、高级消费者、低级消费者和生产者

安装

composer require oemsaas/kafka

示例

• 生产者

    /*
     * 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());
  

• 消费者