介绍
介绍
上一课大家跟福哥学会了在我们的TFLinux系统上面安装Kafka软件，今天福哥要带着大家学习使用PHP去操作Kafka消息系统的方法。
PHP操作Kafka需要借助rdkafka库，我们可以在github上面下载到源代码进行安装。因为rdkafka是作为PHP扩展部署的，所以我们不需要重新编译PHP环境。
安装
下载
下载librdkafka
wget https://github.com/edenhill/librdkafka/archive/v0.11.0.tar.gz
下载php-rdkafka
wget https://github.com/arnaud-lb/php-rdkafka/archive/3.0.4.tar.gz
安装
安装librdkafka
tar -xzvf v0.11.0.tar.gz cd librdkafka-0.11.0/ ./configure --prefix=/tongfu.net/env/librdkafka/ make make install cd ..
安装php-rdkafka
tar -xzvf 3.0.4.tar.gz cd php-rdkafka-3.0.4/ /tongfu.net/env/php-7.4.6/bin/phpize ./configure \ --with-php-config=/tongfu.net/env/php-7.4.6/bin/php-config \ --with-rdkafka=/tongfu.net/env/librdkafka/ make make install cd ..
配置
配置
配置php.ini
extension=rdkafka
重启
重新启动apache
systemctl restart httpd
使用
低级消费者
概念
低级消费者模式就是一个人接了一个项目，所有事情都要自己一点点做。如果当天没有做完，我会记录做到第几件了，然后第二天来了继续做下面的工作。
代码
低级消费者示例代码
当 $message 为 null 或者 $message->err 为到底了，都表示没有新消息
强烈建议本地保存 offset（偏移），我做了多少我自己记着，放心！
$groupID = "lowLevel";
$topicName = "test";
$partitionID = 0;
$conf = new RdKafka\Conf();
$conf->setErrorCb(function($kafka, $err, $reason){
var_dump($err);
var_dump($reason);
});
$conf->set('group.id', $groupID);
$consumer = new RdKafka\Consumer($conf);
$consumer->addBrokers("server_kafka:9092");
$topicConf = new RdKafka\TopicConf();
$topicConf->set('offset.store.method', 'file');
$topicConf->set('offset.store.path', __DIR__);
$topicConf->set('auto.commit.enable', 1);
$topicConf->set('auto.commit.interval.ms', 10);
$topicConf->set('auto.offset.reset', 'smallest');
$topic = $consumer->newTopic($topicName, $topicConf);
$topic->consumeStart($partitionID, RD_KAFKA_OFFSET_STORED);
var_dump("启动中...");
while (1) {
// try consumer record
$message = $topic->consume($partitionID, 120*1000);
if($message == null){
continue;
}
switch ($message->err) {
case RD_KAFKA_RESP_ERR_NO_ERROR:
var_dump($message->payload);
break;
case RD_KAFKA_RESP_ERR__PARTITION_EOF:
var_dump("没有更多消息了");
break;
case RD_KAFKA_RESP_ERR__TIMED_OUT:
var_dump("太长时间未收到消息了");
break;
default:
throw new Exception($message->errstr(), $message->err);
break;
}
}效果
生产一个消息
高级消费者
概念
高级消费者模式就是团队做项目，要做的事情列出来了，你接一个，我接一个，各自做完手头的工作后再去接下一个。分配工作的是 Kafka 的 Rebalance 来控制，逻辑也非常简单。
它先把项目需要做的事情分开多个袋子里，然后每个袋子上写一个编号。当员工来接活的时候，会先看看有没有没人负责的袋子，如果有就告诉员工你的编号是这个，这个袋子里的事情都是你的。如果全部袋子都有人负责了，那就告诉员工暂时没事干，你先等等吧。
注意：消费者能不能拿到消息，完全看是不是可以得到分配到分区，而往往是这个地方会需要等很久~~
代码
高级消费者示例代码
$groupID = "highLevel";
$topicName = "test";
$conf = new RdKafka\Conf();
$conf->setErrorCb(function($kafka, $err, $reason){
var_dump($err);
var_dump($reason);
});
$conf->setRebalanceCb(function (RdKafka\KafkaConsumer $kafka, $err, array $partitions = null) {
switch ($err) {
case RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS:
foreach($partitions as $partition){
var_dump("指定分区：". $partition->getPartition());
}
$kafka->assign($partitions);
break;
case RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS:
foreach($partitions as $partition){
var_dump("删除分区：". $partition->getPartition());
}
$kafka->assign(NULL);
break;
default:
throw new \Exception($err);
}
});
$conf->set('group.id', $groupID);
$conf->set('metadata.broker.list', "127.0.0.1:9092");
$topicConf = new RdKafka\TopicConf();
$topicConf->set('offset.store.method', 'file');
$topicConf->set('offset.store.path', __DIR__);
$topicConf->set('auto.commit.enable', 1);
$topicConf->set('auto.commit.interval.ms', 10);
$topicConf->set('auto.offset.reset', 'smallest');
$conf->setDefaultTopicConf($topicConf);
$consumer = new RdKafka\KafkaConsumer($conf);
$consumer->subscribe(array($topicName));
var_dump("启动中...");
while (1) {
// try consumer record
$message = $consumer->consume(120*1000);
switch ($message->err) {
case RD_KAFKA_RESP_ERR_NO_ERROR:
var_dump($message->payload);
break;
case RD_KAFKA_RESP_ERR__PARTITION_EOF:
var_dump("没有更多消息了");
break;
case RD_KAFKA_RESP_ERR__TIMED_OUT:
var_dump("太长时间未收到消息了");
break;
default:
throw new Exception($message->errstr(), $message->err);
break;
}
}效果
生产一个消息
生产者
代码
生产者示例代码
使用 setErrorCb 回调检查程序是否出错
使用 setDrMsgCb 回调检查消息是否成功推送
设置 message.timeout 避免 produce 阻塞
$conf = new RdKafka\Conf();
$conf->setErrorCb(function($kafka, $err, $reason){
var_dump($err);
var_dump($reason);
});
$conf->setDrMsgCb(function($kafka, $message){
if($message->err) var_dump("发送失败：");
else var_dump("发送成功：");
var_dump($message);
});
$rk = new RdKafka\Producer($conf);
$rk->addBrokers("server_kafka:9092");
$topicConf = new RdKafka\TopicConf();
$topicConf->set("message.timeout.ms", 3000);
$topic = $rk->newTopic("test", $topicConf);
$topic->produce(RD_KAFKA_PARTITION_UA, 0, "福哥说：现在". date("H:i:s"). "了");官方在线文档
Rdkafka扩展提供了在线文档，可以帮助我们编写代码的时候查阅
http://arnaud.le-blanc.net/php-rdkafka/phpdoc/book.rdkafka.html
总结
今天福哥带着童鞋们学习了使用PHP语言通过rdkafka扩展连接操作Kafka消息系统的方法。PHP本身是弱类型语言，又是解析型语言，在处理复杂的业务逻辑时候难免会有些力不从心。借助Kafka消息系统可以将一些同步要求不高的处理放到消息队列里面，可以大大提高主线业务的处理效率。
好了，下一课福哥会给大家讲解如何使用Java语言来操作Kafka消息系统，敬请期待~~