init: 项目初始代码

2026-10-14 14:00:29 +08:00
commit cc434cf89f
464 changed files with 66984 additions and 0 deletions
--- a/app/library/KafkaClient.php
+++ b/app/library/KafkaClient.php
@@ -0,0 +1,151 @@
+<?php
+
+namespace app\library;
+
+/**
+ * Kafka操作类
+ */
+trait KafkaClient
+{
+    /**
+     * 定义Kafka生产者初始化配置
+     * 需要在服务端配置 kafka_server_jaas.conf 里面添加对应的用户名和密码
+     * @var array
+     */
+    private $kafkaProducerConfig =  [];
+
+    /**
+     * 定义Kafka生产者初始化配置
+     * 需要在服务端配置 kafka_client_jaas.conf 里面添加对应的用户名和密码
+     * @var array
+     */
+    private $kafkaConsumerConfig =  [];
+
+
+    /**
+     * 初始化kafka生产配置
+     * @return string[]
+     */
+    private function initKafkaProducerConfig() :array
+    {
+        return $this->kafkaProducerConfig = [
+            'username'      => 'orico',
+            'password'      => 'password',/**/
+            'broker_list'   => '127.0.0.1:9092',/*如果有多个，ip:port,ip2:port,ip3:port*/
+            'topic'         => 'topics'
+        ];
+    }
+
+    /**
+     * 初始化kafka消费配置
+     * @return string[]
+     */
+    private function initKafkaConsumerConfig() :array
+    {
+        return $this->kafkaConsumerConfig = [
+            'username'      => 'orico',
+            'password'      => 'password',/**/
+            'brokerlist'    => '127.0.0.1:9092',/*如果有多个，ip:port,ip2:port,ip3:port*/
+            'topic'         => 'topics'
+        ];
+    }
+
+    /**
+     * kafka 生产者核心方法
+     * @param array $config kafka 配置参数
+     * @param array $data 需要生产的数据
+     */
+    public function kafkaProducer(array $config, array $data)
+    {
+        if (empty($this->kafkaProducerConfig)) {
+            $this->initKafkaProducerConfig();
+        }
+        $conf = new \RdKafka\Conf();
+        $conf->set('sasl.mechanisms', 'PLAIN');
+        $conf->set('security.protocol', 'SASL_PLAINTEXT');
+        $conf->set('sasl.username', $config['username']);
+        $conf->set('sasl.password', $config['password']);
+        $conf->set('socket.keepalive.enable', 'true'); // socket不操作保持长连
+        $conf->set('log.connection.close', 'false'); // 静默所有自发断开连接日志
+        $conf->set('metadata.broker.list', $config['broker_list']);
+
+        $producer = new \RdKafka\Producer($conf);
+
+        $topic = $producer->newTopic($this->topicProduce);
+
+        for ($i = 0; $i < 10; $i++) {
+            $topic->produce(RD_KAFKA_PARTITION_UA, 0,$data);
+            $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!');
+        }
+
+    }
+
+
+    /**
+     * kafka 消息者核心方法
+     * @param array $config kafka 配置参数
+     * @param array $data 需要生产的数据
+     */
+    public function kafkaConsumer(array $config, array $data)
+    {
+        if (empty($this->kafkaConsumerConfig)) {
+            $this->initKafkaConsumerConfig();
+        }
+        return ($this->kafkaConfig());
+    }
+
+    /**
+     * curl请求
+     * @param $url  string 请求的url链接
+     * @param $data string|array|mixed 请求的数据
+     * @param bool $is_post 是否是post请求，默认false
+     * @param array $options 是否附带请求头
+     * @return array|mixed
+     */
+    private function http(string $url, array $data, bool $is_post=false, array $options=[]):array
+    {
+        $data  = json_encode($data);
+        $headerArray = [
+            'Content-type: application/json;charset=utf-8',
+            'Accept: application/json'
+        ];
+        $curl = curl_init();
+        $arr = [];
+        array_push($arr,$url);
+        curl_setopt($curl, CURLOPT_URL, $url);
+        curl_setopt($curl, CURLOPT_SSL_VERIFYPEER, FALSE);
+        curl_setopt($curl, CURLOPT_SSL_VERIFYHOST,FALSE);
+        if ($is_post) {
+            curl_setopt($curl, CURLOPT_POST, 1);
+            curl_setopt($curl, CURLOPT_POSTFIELDS, $data);
+        }
+        curl_setopt($curl,CURLOPT_HTTPHEADER,$headerArray);
+        curl_setopt($curl, CURLOPT_RETURNTRANSFER, 1);
+        if (!empty($options['cookie'])) {
+            curl_setopt($curl, CURLOPT_COOKIE, $options['cookie']);
+        }
+        $output = curl_exec($curl);
+        $http_status = curl_errno($curl);
+        $http_msg = curl_error($curl);
+        curl_close($curl);
+        if ($http_status == 0) {
+            return json_decode($output, true);
+        } else {
+            return ['status' => $http_status, 'message' => $http_msg, 'data' => []];
+        }
+    }
+
+
+
+}