在 PHP 中使用 Kafka：库、安装和代码示例

在 PHP 中使用 Kafka：库概览与实践示例

Apache Kafka 是一个分布式流数据处理平台，已成为构建高负载消息交换系统的事实标准。在 PHP 生态系统中，可以通过多个库实现与 Kafka 的集成，每个库都有其自身特点。本文将介绍两个主要库：php-rdkafka（基于 C 的扩展）和 longlang/phpkafka（纯 PHP），并展示生产者和消费者的代码示例。

1. php-rdkafka 库

php-rdkafka 是一个基于 librdkafka 库的原生 PHP 扩展。它提供最佳性能和完整的 Kafka 配置控制，适用于高负载的生产项目。

安装

安装需要 librdkafka 和 PECL：

# 安装 librdkafka (Ubuntu/Debian)sudo apt-get install librdkafka-dev

# 通过 PECL 安装扩展sudo pecl install rdkafka

# 在 php.ini 中启用extension=rdkafka.so

示例：生产者 (Producer)

生产者将消息发送到 Kafka 主题。正确配置异步发送和错误处理非常重要：

<?phpuse RdKafka\\Producer;use RdKafka\\Conf;

$conf = new Conf();$conf->set('bootstrap.servers', 'localhost:9092');$conf->set('acks', 'all'); // 等待所有副本确认

$producer = new Producer($conf);$topic = $producer->newTopic('test-topic');

// 发送消息$topic->produce(RD_KAFKA_PARTITION_UA, 0, 'Hello Kafka!');$producer->flush(1000); // 等待投递（1秒）

echo "消息已发送";

示例：消费者 (Consumer)

消费者从主题中读取消息。建议使用消费者组来实现负载均衡：

<?phpuse RdKafka\\KafkaConsumer;use RdKafka\\Conf;

$conf = new Conf();$conf->set('bootstrap.servers', 'localhost:9092');$conf->set('group.id', 'my-consumer-group');$conf->set('auto.offset.reset', 'earliest');

$consumer = new 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;        default:            echo "错误: " . $message->errstr() . "";            break;    }}

2. longlang/phpkafka 库

longlang/phpkafka 是一个纯 PHP 实现的 Kafka 协议库，无外部依赖。它安装更简单（无需编译），适用于开发、测试或性能要求不高的项目。

安装

通过 Composer 安装：

composer require longlang/phpkafka

示例：生产者

<?phpuse longlang\\phpkafka\\Producer\\Producer;use longlang\\phpkafka\\Producer\\ProducerConfig;

$config = new ProducerConfig();$config->setBootstrapServer('localhost:9092');$config->setUpdateBrokers(true);

$producer = new Producer($config);$producer->send('test-topic', 'Hello Kafka from pure PHP!', 'key-1');echo "消息已发送";

示例：消费者

<?phpuse longlang\\phpkafka\\Consumer\\Consumer;use longlang\\phpkafka\\Consumer\\ConsumerConfig;

$config = new ConsumerConfig();$config->setBootstrapServer('localhost:9092');$config->setGroupId('my-group');$config->setTopic('te