lumen代码
lass MsgController extends Controller
{
public function test():void {
$rk = new \RdKafka\Producer();
//$rk->setLogLevel(LOG_DEBUG);
$rk->addBrokers("127.0.0.1");
$topic = $rk->newTopic('test');
for ($i=0;$i<=1000000;$i++){
$json = json_encode(['a'=>$i]);
$topic->produce(RD_KAFKA_PARTITION_UA, 0, $json);
}
}
}
查看消费
kafka-console-consumer -bootstrap-server localhost:9092 -topic test -from-beginning
理应如图:
WechatIMG407.jpeg
网友评论