如何使用PHP的php-rdkafka库获取Kafka主题的最后偏移量？

偏移量返回-1001的原因 #

你用getOffsetPositions()拿到的-1001对应常量RD_KAFKA_OFFSET_INVALID，这个方法的作用是获取消费者自身已存储/提交的偏移量位置，而非主题的最新消息偏移量。当消费者还未分配分区、从未提交过偏移量时，就会返回这个无效值。

正确获取主题最后偏移量的方法 #

要获取主题分区的最新偏移量，需使用getWatermarkOffsets()方法，它会返回分区的最低（最早）和最高（下一个待写入）偏移量，其中最高偏移量减1就是当前最后一条消息的偏移量：

// 补充初始化消费者实例（你原代码中遗漏了这一步）
$consumer = new RdKafka\Consumer($conf);

$topicName = 'userstatistics_12345';
$partition = 0;

// 获取水位线偏移量
$lowOffset = $highOffset = 0;
$err = $consumer->getWatermarkOffsets($topicName, $partition, $lowOffset, $highOffset);
if ($err !== RD_KAFKA_RESP_ERR_NO_ERROR) {
    throw new Exception(RdKafka\err2str($err));
}

// 最后一条消息的偏移量为最高偏移量减1
$lastOffset = $highOffset - 1;
echo "当前主题最后偏移量：{$lastOffset}\n";

指定偏移量开始消费 #

拿到正确的$lastOffset后，可按你原本的思路指定消费起始位置：

$consumer->assign([
    new RdKafka\TopicPartition($topicName, $partition, $lastOffset)
]);

避免while(true)的高效消费方案 #

不要用无限制死循环，可通过带超时的轮询+优雅退出条件实现高效消费，比如设置轮询超时时间、监听退出信号：

// 注册信号处理，支持Ctrl+C优雅退出
pcntl_signal(SIGINT, function() use (&$running) {
    $running = false;
});

$running = true;
while ($running) {
    // 处理信号
    pcntl_signal_dispatch();
    
    // 轮询消息，设置超时时间（示例为1000ms，即1秒）
    $message = $consumer->consume(1000);
    
    switch ($message->err) {
        case RD_KAFKA_RESP_ERR_NO_ERROR:
            // 处理消息并发送到API
            // sendToApi($message->payload);
            
            // 手动提交偏移量（因你关闭了自动提交）
            $consumer->commit($message);
            break;
        case RD_KAFKA_RESP_ERR__TIMED_OUT:
            // 超时无消息，可在此执行辅助逻辑或直接跳过
            break;
        default:
            throw new Exception(RdKafka\err2str($message->err));
    }
}

// 消费结束后清理资源
$consumer->close();

这种方式既不会持续占用CPU（超时后释放资源），又能及时响应新消息，同时支持优雅退出，比无限制while(true)更高效可控。

内容的提问来源于stack exchange，提问作者Anton