在 PHP 中使用 Kafka：库、连接和代码示例

在 PHP 中使用 Kafka：库、连接和代码示例

Apache Kafka 是一个分布式流数据平台，可以实时发布、存储和处理消息流。在 PHP 生态系统中，使用 php-rdkafka 扩展来与 Kafka 交互，该扩展是用 C 语言编写的 librdkafka 库的封装。在本文中，我们将介绍如何安装该工具、编写生产者和消费者，并探讨典型的使用场景。

什么是 php-rdkafka？

php-rdkafka 是用于 Apache Kafka 的官方 PHP 扩展。它提供了一个高性能的接口，用于发送和接收消息。该扩展支持 Kafka 的所有关键功能：主题、分区、消费者组、手动提交、压缩等。

主要功能：

• 创建生产者 (producers) 以向主题发送消息。
• 创建消费者 (consumers) 以从主题读取消息。
• 支持消费者组和自动再平衡。
• 处理分区和手动管理偏移量。
• 消息压缩（gzip、snappy、lz4、zstd）。
• 异步发送和批量处理。

安装 php-rdkafka

可以通过 PECL 或从源码编译来安装该扩展。推荐的方式是通过 PECL：

# 通过 PECL 安装pecl install rdkafka

# 对于 Ubuntu/Debian，请先安装 librdkafka-devsudo apt-get install librdkafka-devpecl install rdkafka

# 在 php.ini 中添加扩展extension=rdkafka.so

安装后，检查扩展是否已加载：

php -m | grep rdkafka

如果您使用 Composer，还可以额外安装 arnaud-lb/php-rdkafka 包以实现自动加载：

composer require arnaud-lb/php-rdkafka

生产者示例 (Producer)

生产者将消息发送到指定的 Kafka 主题。我们来看一个同步发送的基本示例：

<?phprequire_once 'vendor/autoload.php';

$conf = new RdKafka\\Conf();$conf->set('metadata.broker.list', 'localhost:9092');

$producer = new RdKafka\\Producer($conf);$topic = $producer->newTopic('test_topic');

// 发送消息$topic->produce(RD_KAFKA_PARTITION_UA, 0, 'Hello, Kafka from PHP!');

// 等待所有消息发送完成$producer->flush(1000);echo "消息已发送";

代码说明：

• metadata.broker.list — Kafka 代理地址（默认为 localhost:9092）。
• RD_KAFKA_PARTITION_UA — 自动选择分区。
• flush(1000) — 阻塞执行直到所有消息发送完成（超时时间 1 秒）。

消费者示例 (Consumer)

消费者从主题中读取消息。需要指定消费者组和偏移量提交策略：

<?phprequire_once 'vendor/autoload.php';

$conf = new RdKafka\\Conf();$conf->set('metadata.broker.list', 'localhost:9092');$conf->set('group.id', 'my_consumer_group');$conf->set('auto.offset.reset', 'earliest');

$consumer = new RdKafka\\KafkaConsumer($conf);$consumer->subscribe(['test_topic']);

echo "等待消息...";

while (true) {    $message = $consumer->consume(120 * 1000); // 超时时间 2 分钟    switch ($message->err) {        case RD_KAFKA_RESP_ERR_NO_ERROR:            echo "收到消息: " . $message->payload . "";            break;        case RD_KAFKA_RESP_ERR__PARTITION_EOF:            echo "已到达分区末尾";            break;        case RD_KAFKA_RESP_ERR__TIMED_OUT:            echo "超时";            break;        defau