php使用kafka

1:安装扩展

2：生产者生产消息

$rk = new \RdKafka\Producer();
$rk->addBrokers("localhost:9092"); //kafka服务器地址
self::$producer = $rk;
$topic = self::$producer->newTopic($topic);
$value ='这是消息';
$topic->produce(RD_KAFKA_PARTITION_UA, 0, $value);
self::$producer->flush(1000); // 等待消息发送完成

3：消费者消费

$conf = new \RdKafka\Conf();
$conf->set('group.id', 'my_consumer_group');
$conf->set('bootstrap.servers', 'localhost:9092');
$rk = new \RdKafka\KafkaConsumer($conf);
self::$consume = $rk;
self::$consume->subscribe($topic);
while (true) {
$message = self::$consume->consume(1000);
switch ($message->err) {
case RD_KAFKA_RESP_ERR_NO_ERROR:
echo '接收到消息: ' . $message->payload . PHP_EOL;
break;
case RD_KAFKA_RESP_ERR__PARTITION_EOF:
echo '已达到分区末尾，没有更多消息可消费' . PHP_EOL;
break;
case RD_KAFKA_RESP_ERR__TIMED_OUT:
// echo '消费超时' . PHP_EOL;
break;
default:
echo '消费错误: ' . $message->errstr() . PHP_EOL;
break;
}
}