PHP处理Kafka数据怎么读写-php教程-PHP中文网

rdkafka是php唯一稳定、活跃、支持sasl/ssl/exactly-once的生产级kafka扩展，需手动安装librdkafka和rdkafka扩展，禁用自动提交offset并用cli常驻进程运行消费者。

php处理kafka数据怎么读写

PHP读Kafka用rdkafka扩展，不是php-kafka或kafka-php

PHP原生不支持Kafka，必须依赖C扩展。rdkafka是目前唯一稳定、活跃、支持SASL/SSL/Exactly-Once语义的生产级扩展，php-kafka和kafka-php都是纯PHP实现，吞吐低、不维护、不兼容新协议，线上别碰。

安装方式（Ubuntu/Debian）：

sudo apt-get install librdkafka1-dev
pecl install rdkafka
echo "extension=rdkafka.so" >> /etc/php/*/cli/php.ini
echo "extension=rdkafka.so" >> /etc/php/*/fpm/php.ini

验证是否加载成功：

php -m | grep rdkafka —— 有输出即为成功

注意：PHP 8.1+ 需用 rdkafka ≥6.0.1；若用Confluent Cloud，必须启用sasl.mechanisms=PLAIN和security.protocol=sasl_ssl，否则连不上。

消费者怎么拉取并正确提交offset

rdkafka消费者默认是自动提交（enable.auto.commit=true），但实际中容易丢数据——比如消息处理到一半进程崩溃，offset已提交，下次启动就跳过这条。

推荐手动控制提交时机，示例关键逻辑：

$conf = new \RdKafka\Conf();
$conf->set('group.id', 'my-group');
$conf->set('auto.offset.reset', 'latest');
$conf->set('enable.auto.commit', 'false'); // 关键：关掉自动提交
$conf->set('bootstrap.servers', 'kafka-broker:9092');

$consumer = new \RdKafka\Consumer($conf);
$consumer->subscribe(['my-topic']);

while (true) {
    $message = $consumer->consume(1000); // 1s超时
    if ($message->err === RD_KAFKA_RESP_ERR_NO_ERROR) {
        // 处理业务逻辑（这里不能抛异常！）
        process_message($message->payload);

        // 处理成功后才提交当前offset
        $consumer->commit($message);
    } elseif ($message->err === RD_KAFKA_RESP_ERR__PARTITION_EOF) {
        continue; // 到分区尾部，继续轮询
    } else {
        throw new \Exception($message->errstr());
    }
}

常见坑：

commit()传$message只提交该条，传null会提交所有已拉取但未提交的offset，慎用
不要在process_message()里sleep或阻塞太久，否则触发session.timeout.ms（默认10s），消费者被踢出group
如果处理失败需重试，不要commit()，下次consume()还会拿到同一条

生产者发消息要注意ack和重试配置

默认acks=1（leader写入即返回），但网络抖动或leader切换时可能丢数据。关键配置项：

btpanel phpsite 宝塔面板PHP网站

宝塔面板 PHP 网站管理：站点创建、删除、启停、PHP 版本切换、域名管理、SSL证书管理、伪静态管理、数据库管理

下载

acks=all：等ISR中所有副本写入才返回，最安全，但延迟略高
retries=MAX_INT：rdkafka内部自动重试，别自己封装for循环重发
retry.backoff.ms=100：重试间隔，避免高频打满broker
message.timeout.ms=30000：单条消息从发送到确认的总时限，超时会进error_cb

基础发送示例：

$conf = new \RdKafka\Conf();
$conf->set('bootstrap.servers', 'kafka-broker:9092');
$conf->set('acks', 'all');
$conf->set('retries', '2147483647');
$conf->set('retry.backoff.ms', '100');
$conf->set('message.timeout.ms', '30000');

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

// 发送（异步，立即返回）
$topic->produce(RD_KAFKA_PARTITION_UA, 0, 'my-message', null, 0);

// 强制刷出缓冲区（可选，调试用）
$producer->flush(1000);

注意：produce()不抛异常，错误走error_cb回调，必须设：

$conf->setErrorCb(function ($kafka, $err, $reason) { error_log("Kafka error: $err $reason"); });