<?php

try {
    $rk = new \RdKafka\Consumer();
    //$rk->setLogLevel(LOG_DEBUG); // 设置日志级别
    $rk->addBrokers("49.234.210.15:9092"); // 添加经纪人，就是ip地址

    $topic = $rk->newTopic("topic"); // 这里的$rk和生产者是不同的类哦
    // 第一个参数分区ID
    // 第二个参数是开始消费的偏移量，有效值
    $topic->consumeStart(0, RD_KAFKA_OFFSET_BEGINNING);
    $true = true;
    while ($true) {
        // 第一个参数是要消耗的分区
        // 第二个参数是等待收到消息的最长时间，1000是一秒
        $msg = $topic->consume(0, 1000);
        if ($msg->err) {
            echo __LINE__.$msg->errstr(), "\n";
            break;
        } else {
            echo __LINE__.$msg->payload, "\n";
            if (!$msg->payload){
                $true = false;
            }
        }
    }
} catch (\Exception $e) {
    var_dump($e->getLine() . ' | ' . $e->getMessage());
}