安装rdkafka
rdkafka 依赖 libkafka
yum install rdkafka rdkafka-devel
pecl install rdkafka
php --ri rdkafka
http://pecl.php.net/package/r… 可以参阅支持的kafka客户端版本
生产者
连接集群，创建 topic，生产数据。
<?php
$rk = new Rdkafka\Producer();
$rk->setLogLevel(LOG_DEBUG);
// 链接kafka集群
$rk->addBrokers("192.168.20.6:9092,192.168.20.6:9093");
// 创建topic
$topic = $rk->newTopic("topic_1");
while (true) {
$message = "hello kafka " . date("Y-m-d H:i:s");
echo "hello kafka " . date("Y-m-d H:i:s") . PHP_EOL;
try {
$topic->produce(RD_KAFKA_PARTITION_UA, 0, $message);
sleep(2);
} catch (\Exception $e) {
echo $e->getMessage() . PHP_EOL;
}
}
消费者-HighLevel
自动分配partition，rebalance，comsumer group。
<?php
$conf = new RdKafka\Conf();
// Set a rebalance callback to log partition assignments (optional)
$conf->setRebalanceCb(function (RdKafka\KafkaConsumer $kafka, $err, array $partitions = null) {
switch ($err) {
case RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS:
echo "Assign: ";
var_dump($partitions);
$kafka->assign($partitions);
break;
case RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS:
echo "Revoke: ";
var_dump($partitions);
$kafka->assign(null);
break;
default:
throw new \Exception($err);
}
});
// Configure the group.id. All consumer with the same group.id will consume
// different partitions.
$conf->set('group.id', 'group_1');
// Initial list of Kafka brokers
$conf->set('metadata.broker.list', '192.168.20.6:9092,192.168.20.6:9093');
$topicConf = new RdKafka\TopicConf();
// Set where to start consuming messages when there is no initial offset in
// offset store or the desired offset is out of range.
// 'smallest': start from the beginning
$topicConf->set('auto.offset.reset', 'smallest');
// Set the configuration to use for subscribed/assigned topics
$conf->setDefaultTopicConf($topicConf);
$consumer = new RdKafka\KafkaConsumer($conf);
// Subscribe to topic 'topic_1'
$consumer->subscribe(['topic_1']);
echo "Waiting for partition assignment... (make take some time when\n";
echo "quickly re-joining the group after leaving it.)\n";
while (true) {
$message = $consumer->consume(3e3);
switch ($message->err) {
case RD_KAFKA_RESP_ERR_NO_ERROR:
var_dump($message);
break;
case RD_KAFKA_RESP_ERR__PARTITION_EOF:
sleep(2);
case RD_KAFKA_RESP_ERR__TIMED_OUT:
echo $message->errstr() . PHP_EOL;
break;
default:
throw new \Exception($message->errstr(), $message->err);
break;
}
}
消费者-LowLevel
指定partition消费。
php consumer_lowlevel.php [partitonNuo]
LowLevel 没有消费组的概念，也可以认为每个消费者都属于一个独立消费组。
<?php
if (!isset($argv[1])) {
fwrite(STDOUT, "请指定消费分区：");
$partition = (int) fread(STDIN, 1024);
} else {
$partition = (int) $argv[1];
}
$topic = "topic_1";
$conf = new RdKafka\Conf();
// Set the group id. This is required when storing offsets on the broker
$conf->set('group.id', 'group_2');
$rk = new RdKafka\Consumer($conf);
$rk->addBrokers('192.168.20.6:9092,192.168.20.6:9093');
$topicConf = new RdKafka\TopicConf();
$topicConf->set('auto.commit.interval.ms', 2000);
// Set the offset store method to 'file'
// $topicConf->set('offset.store.method', 'file');
// $topicConf->set('offset.store.path', sys_get_temp_dir());
// Alternatively, set the offset store method to 'broker'
$topicConf->set('offset.store.method', 'broker');
// Set where to start consuming messages when there is no initial offset in
// offset store or the desired offset is out of range.
// 'smallest': start from the beginning
$topicConf->set('auto.offset.reset', 'smallest');
$topic = $rk->newTopic($topic, $topicConf);
// Start consuming partition 0
$topic->consumeStart($partition, RD_KAFKA_OFFSET_STORED);
while (true) {
$message = $topic->consume($partition, 3 * 1000);
switch ($message->err) {
case RD_KAFKA_RESP_ERR_NO_ERROR:
var_dump($message);
break;
case RD_KAFKA_RESP_ERR__PARTITION_EOF:
case RD_KAFKA_RESP_ERR__TIMED_OUT:
echo $message->errstr() . PHP_EOL;
break;
default:
throw new \Exception($message->errstr(), $message->err);
break;
}
}
为您推荐与 kafka, rdkafka 相关的帖子：
暂无回复。
