当前位置：首页 > article >正文

php怎么连接使用kafka

article 2024/9/23 4:59:52

PHP 连接并使用 Kafka 需要借助 Kafka 的 PHP 客户端库，比如流行的 php-rdkafka 扩展。它是基于 C 语言的 librdkafka 库的 PHP 绑定，功能稳定且性能高。下面是如何使用 php-rdkafka 来连接和使用 Kafka 的步骤。

1. 安装 `php-rdkafka`

1.1 安装依赖

首先，你需要安装 librdkafka 库。在 Linux 系统上，可以通过包管理器来安装：

Debian/Ubuntu:
```
sudo apt-get install -y librdkafka-dev
```
CentOS/RHEL:
```
sudo yum install -y librdkafka-devel
```

1.2 安装 `php-rdkafka` 扩展

可以通过以下几种方式安装 php-rdkafka 扩展。

通过 PECL 安装：
```
sudo pecl install rdkafka
```
安装完成后，需要将 extension=rdkafka.so 添加到 PHP 配置文件中，通常位于 /etc/php/7.x/cli/php.ini 或 /etc/php/7.x/apache2/php.ini 中。
```
extension=rdkafka.so
```
验证安装成功：

你可以通过以下命令验证扩展是否安装成功：
```
php -m | grep rdkafka
```
如果显示 rdkafka，说明安装成功。

2. 编写 PHP 代码使用 Kafka

安装好扩展后，你可以编写 PHP 脚本来生产和消费 Kafka 消息。下面是基本的生产者和消费者示例。

2.1 Kafka 生产者代码示例

生产者用于将消息发送到 Kafka。

<?php

// 创建 Kafka 配置
$config = new RdKafka\Conf();

// 设置生产者配置参数（可根据需求自定义）
$config->set('metadata.broker.list', 'localhost:9092');

// 创建生产者实例
$producer = new RdKafka\Producer($config);

if (!$producer) {
    die("Failed to create producer\n");
}

// 创建一个生产主题
$topic = $producer->newTopic("test");

// 生产消息并发送到 Kafka
for ($i = 0; $i < 10; $i++) {
    $message = "Message $i";
    $topic->produce(RD_KAFKA_PARTITION_UA, 0, $message);
    echo "Produced: $message\n";
}

// 刷新缓冲区，将消息真正发送到 Kafka
$producer->flush(10000);

2.2 Kafka 消费者代码示例

消费者用于从 Kafka 中接收消息。

<?php

// 创建一个 RdKafka 配置对象
$conf = new RdKafka\Conf();

// 设置消费者组 ID 为 'myConsumerGroup'。这是必须的，当在 broker 上存储偏移量时需要它。
$conf->set('group.id', 'myConsumerGroup');

// 当到达分区的末尾时发出 EOF（文件结束）事件。
$conf->set('enable.partition.eof', 'true');

// 使用配置创建一个新的 RdKafka 消费者实例
$rk = new RdKafka\Consumer($conf);

// 将本地 broker 的 IP 地址添加到消费者的 broker 列表中。这里假设 broker 运行在本机上。
$rk->addBrokers("192.168.3.20");

// 创建一个新的主题配置对象
$topicConf = new RdKafka\TopicConf();

// 设置自动提交偏移量的时间间隔为 100 毫秒
$topicConf->set('auto.commit.interval.ms', 1000);

// 设置偏移量存储方法为 'broker'，意味着偏移量将被存储在 Kafka broker 上
$topicConf->set('offset.store.method', 'broker');

// 设置当没有初始偏移量或所需偏移量超出范围时从最早的消息开始消费
$topicConf->set('auto.offset.reset', 'earliest');

// 使用主题配置创建一个名为 "test" 的新主题
$topic = $rk->newTopic("test", $topicConf);

// 开始消费主题 "test" 的分区 0，从已存储的偏移量处开始
$topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);

// 进入一个无限循环，不断地尝试从分区 0 获取消息
while (true) {
    // 尝试消费一条消息，并设置超时时间为 120 秒（120*10000 毫秒）
    $message = $topic->consume(0, 120 * 1000); // 注意：这里的单位是毫秒，所以应该是 120*1000 而不是 120*10000
    var_dump($message);
    // 根据收到的消息错误码进行处理
    switch ($message->err) {
        case RD_KAFKA_RESP_ERR_NO_ERROR:
            // 如果没有错误，则打印消息内容
            var_dump($message);
            break;
        case RD_KAFKA_RESP_ERR__PARTITION_EOF:
            // 如果达到了分区的末尾，则输出提示信息表示没有更多消息可消费
            echo "No more messages; will wait for more\n";
            break;
        case RD_KAFKA_RESP_ERR__TIMED_OUT:
            // 如果超时，则输出超时信息
            echo "Timed out\n";
            break;
        default:
            // 对于其他错误情况，则抛出异常
            throw new \Exception($message->errstr(), $message->err);
            break;
    }
}

?>

3. 配置说明

生产者配置（Producer）：
- metadata.broker.list：Kafka broker 地址，通常为 localhost:9092（或者根据实际情况设置）。
消费者配置（KafkaConsumer）：
- group.id：消费者组 ID，用于标识消费者组。

4. 运行 PHP Kafka 代码

确保 Kafka 正在运行。

运行生产者：

通过命令行运行 PHP 脚本：
```
php kafka_producer.php
```
你会看到消息被生产并发送到 Kafka。
运行消费者：

启动消费者脚本：
```
php kafka_consumer.php
```
你会看到消费者从 Kafka 接收并处理消息。

5. 调试和常见问题

生产者或消费者连接问题：确保 Kafka 的 broker 地址正确。如果 Kafka 运行在 Docker 容器中，可能需要使用容器的 IP 地址或配置 KAFKA_ADVERTISED_LISTENERS。
超时问题：如果消费者长时间没有收到消息，可以检查消费者组的配置和 Kafka 的主题分区设置。