kafka擴展安裝好后就可以編程測試了，以下代碼是在thinkphp5上實現的。 測試kafka生產者 - ~~~ //測試kafka生產者 public function testProducer() { $broker_list = '127.0.0.1:9092'; $topic = 'test'; $partition = 0; $rk = new \RdKafka\Producer(); $rk->setLogLevel(LOG_DEBUG); $rk->addBrokers($broker_list); $topic = $rk->newTopic($topic); $message = 'this is a php test'; $topic->produce(RD_KAFKA_PARTITION_UA,$partition,json_encode($message)); } ~~~ 測試kafka消費者 - ~~~ //測試kafka消費者 public function testConsumer($object,$callback) { $broker_list = '127.0.0.1:9092'; $topic = 'test'; $conf = new \RdKafka\Conf(); $conf->set('group.id',0); $conf->set('metadata.broker.list',$broker_list); $topicConf = new \RdKafka\TopicConf(); $topicConf->set('auto.offset.reset','smallest'); $conf->setDefaultTopicConf($topicConf); $consumer = new \RdKafka\KafkaConsumer($conf); $consumer->subscribe([$topic]); echo 'waiting message....'; while (true){ $message = $consumer->consume(120*1000); switch ($message->err){ case RD_KAFKA_RESP_ERR_NO_ERROR: echo 'msg payload....'; $object->$callback($message->payload); break; } sleep(1); } } ~~~