袁来如此的工作笔记
袁来如此的工作笔记
竹杖芒鞋轻胜马，谁怕？ 一蓑烟雨任平生。

Thinkphp RdKafka 公共类封装

浏览量:92

在 extend/RdKafkaClient.php 封装 RdKafkaClient：

<?php
 
ini_set('default_socket_timeout', -1);
use RdKafka;
 
class RdKafkaClient
{
 
    public static $instance = null; //
    //单例模式
    private $base_config;
    private $kafka_conf;
    private $kafka_topic_conf;
 
    private $producer;
    private $consumer;
 
 
    /**
     * 初始化kafka
     * $config = array(
     *  'server' => '127.0.0.1' 服务器
     *  'port'   => '6379' 端口号
     * )
     * @param array $config
     * @throws Exception
     */
    public function __construct()
    {
        try {
            // 读取tp的配置文件
            $this->base_config = config('game.kafka') ?? [];
 
            // 实例化Conf
            $this->kafka_conf = new Rdkafka\Conf();
            
            // 实例化topicConf
            $topicConf = new RdKafka\TopicConf();
            $topicConf->set('auto.commit.interval.ms', 100);
            $topicConf->set('offset.store.method', 'broker');
            $topicConf->set('auto.offset.reset', 'earliest');
 
            $this->kafka_topic_conf = $topicConf;
        } catch (\Exception $e) {
            error_log($e->getMessage());
            throw new Exception($e->getMessage());
        }
 
        return true;
    }
 
    /**
     * 新增单例模式
     * @param array $config
     * @return static
     * @throws Exception
     */
    public static function getInstance()
    {
        $key = md5(json_encode(array()));
        if (!isset(self::$instance[$key])) {
            self::$instance[$key] = new self();
        }
        return self::$instance[$key];
    }
 
    public function getConf() {
        $this->kafka_conf->set('metadata.broker.list', $this->base_config['brokers']);
        return $this->kafka_conf; 
    }
 
    public function getProducer() {
        return $this->producer = new RdKafka\Producer($this->kafka_conf);
    }
 
    public function getConsumer($group_id) {
        $this->kafka_conf->set('group.id', $group_id);
 
        $this->consumer = new RdKafka\Consumer($this->kafka_conf);
        $this->consumer->addBrokers($this->base_config['brokers']);
        return $this->consumer;
    }
 
    public function setConsumeTopic($topic) {
        $topic = $this->consumer->newTopic($topic, $this->kafka_topic_conf);
 
        $topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);
 
        return $topic;
    }
 
    /**
     * 调用方法不存在,就直接return false,防止报错
     */
    public function __call($name, $arguments)
    {
        $logFile = dirname(__DIR__) . "/data/logs/rdkafka_method_not_exist.log";
        if (file_exists(dirname($logFile))) {
            file_put_contents($logFile, date("Y-m-d H:i:s") . $name . "\n", FILE_APPEND);
        }
        return false;
    }
 
}
 
?>
生产者调用方法：

public function rd_producer() {
    $kafka = RdKafkaClient::getInstance();
 
    $producer = $kafka->getProducer();
 
    $topic = $producer->newTopic("mytopic");
 
    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 rd_consumer() {
    $kafka = RdKafkaClient::getInstance();
    $kafka->getConsumer("aaa");
    $topic = $kafka->setConsumeTopic("mytopic");
 
    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 "No more messages; will wait for more\n";
                break;
            case RD_KAFKA_RESP_ERR__TIMED_OUT:
                echo "Timed out\n";
                break;
            default:
                throw new \Exception($message->errstr(), $message->err);
                break;
        }
    }
}
打赏

ThinkPHP