消息中间件Kafka（PHP版本） - ToB企服应用市场:ToB评测及商务社交产业平台

class Kafka extends CI_Controller {
//定义变量(分区)
private $borker_list = "";
//定义变量（配置）
private $conf = "";
//定义变量（主题）
private $topics = "";
//定义变量（分组）
private $topics_group = "";
//构造
public function __construct(){
parent::__construct();
//初始化数据
$this->borker_list = "localhost:9092";
$this->topics = "test";
$this->topics_group = "test-group";
}
//消息生产者
public function producter(){
//初始化
$conf= new RdKafka\Conf();
//设置分区
$conf->set('metadata.broker.list', $this->borker_list);
//初始化生产者
$producer = new RdKafka\Producer($conf);
//设置主题
$topic = $producer->newTopic($this->topics);
//产生信息
for ($i = 0; $i < 10; $i++) {
$topic->produce(RD_KAFKA_PARTITION_UA, 0, "Message $i");
$producer->poll(0);
}
//消息刷新
for ($flushRetries = 0; $flushRetries < 10; $flushRetries++) {
$result = $producer->flush(10000);
if (RD_KAFKA_RESP_ERR_NO_ERROR === $result) {
break;
}
}
//刷新结果
if (RD_KAFKA_RESP_ERR_NO_ERROR !== $result) {
throw new \RuntimeException('Was unable to flush, messages might be lost!');
}
}
//消息订阅者
public function consumer(){
set_time_limit(0);
//初始化
$conf = new RdKafka\Conf();
//设置分区
$conf->set('metadata.broker.list', $this->borker_list);
$conf->set('group.id',$this->topics_group);
//初始化消费者
$rk = new RdKafka\Consumer($conf);
//主题配置
$topicConf = new RdKafka\TopicConf();
$topicConf->set('auto.commit.interval.ms', 100);
$topicConf->set('offset.store.method', 'file');
$topicConf->set('offset.store.path', sys_get_temp_dir());
$topicConf->set('auto.offset.reset', 'smallest');
$topic = $rk->newTopic($this->topics, $topicConf);
$topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);
while (true) {
$message = $topic->consume(0, 120*10000);
switch ($message->err) {
case RD_KAFKA_RESP_ERR_NO_ERROR:
//没有错误打印信息
var_dump($message);
break;
case RD_KAFKA_RESP_ERR__PARTITION_EOF:
echo "等待接收信息\n";
break;
case RD_KAFKA_RESP_ERR__TIMED_OUT:
echo "超时\n";
break;
default:
throw new \Exception($message->errstr(), $message->err);
break;
}
}
}
//获取元数据（包括主题（topics）、分区（partitions）、副本（replicas）和ISR（In-Sync Replicas）等信息）
public function gettest(){
//初始化
$conf= new RdKafka\Conf();
//设置分区
$conf->set('metadata.broker.list', $this->borker_list);
//初始化生产者
$producer = new RdKafka\Producer($conf);
$topic = $producer->newTopic($this->topics);
// $result = $producer->getMetadata(false, $topic, 10*1000);
$result = $producer->getOutQLen();
var_dump($result);die;
}
//获取元数据
public function metadata(){
$conf = new RdKafka\Conf();
$conf->setDrMsgCb(function ($kafka, $message) {
file_put_contents("./xx.log", var_export($message, true), FILE_APPEND);
});
$conf->setErrorCb(function ($kafka, $err, $reason) {
printf("Kafka error: %s (reason: %s)\n", rd_kafka_err2str($err), $reason);
});
$conf->set('group.id', 'myConsumerGroup');
$rk = new RdKafka\Consumer($conf);
$rk->addBrokers("127.0.0.1");
$allInfo = $rk->getMetadata(true, NULL, 60e3);
$topics = $allInfo->getTopics();
//循环输出
foreach ($topics as $topic) {
$topicName = $topic->getTopic();
if ($topicName == "__consumer_offsets") {
continue ;
}
$partitions = $topic->getPartitions();
foreach ($partitions as $partition) {
$topPartition = new RdKafka\TopicPartition($topicName, $partition->getId());
echo "当前的话题：" . ($topPartition->getTopic()) . " - " . $partition->getId() . " - ";
echo "offset：" . ($topPartition->getOffset()) . PHP_EOL;
}
}
}
}

复制代码