diff --git a/.idea/jarRepositories.xml b/.idea/jarRepositories.xml
index 22e83c6b398d40fa937107583781f65c9e1262b6..ac602a38f492e3418ae2b08f7914a0b6126705ff 100644
--- a/.idea/jarRepositories.xml
+++ b/.idea/jarRepositories.xml
@@ -21,5 +21,10 @@
       <option name="name" value="JBoss Community repository" />
       <option name="url" value="https://repository.jboss.org/nexus/content/repositories/public/" />
     </remote-repository>
+    <remote-repository>
+      <option name="id" value="central" />
+      <option name="name" value="Central Repository" />
+      <option name="url" value="https://maven.aliyun.com/nexus/content/groups/public" />
+    </remote-repository>
   </component>
 </project>
\ No newline at end of file
diff --git a/.idea/misc.xml b/.idea/misc.xml
index 132404bc29da2535f7f7dde84180d85abc60d5a4..5d8d5c0752646929003e045116450ab7cda64718 100644
--- a/.idea/misc.xml
+++ b/.idea/misc.xml
@@ -7,6 +7,7 @@
         <option value="$PROJECT_DIR$/pom.xml" />
       </list>
     </option>
+    <option name="workspaceImportForciblyTurnedOn" value="true" />
   </component>
   <component name="ProjectRootManager" version="2" languageLevel="JDK_1_8" default="true" project-jdk-name="1.8" project-jdk-type="JavaSDK">
     <output url="file://$PROJECT_DIR$/out" />
diff --git a/pom.xml b/pom.xml
index b312a99b2f8b8563691bc940a349f215604c69e9..2d8e66d47c79f264c030a91e8528a02108707a25 100644
--- a/pom.xml
+++ b/pom.xml
@@ -51,6 +51,31 @@
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-data-redis</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>druid-spring-boot-starter</artifactId>
+            <version>1.1.21</version>
+        </dependency>
+        <dependency>
+            <groupId>ru.yandex.clickhouse</groupId>
+            <artifactId>clickhouse-jdbc</artifactId>
+            <version>0.2.4</version>
+        </dependency>
+        <dependency>
+            <groupId>com.baomidou</groupId>
+            <artifactId>mybatis-plus-boot-starter</artifactId>
+            <version>3.5.1</version>
+        </dependency>
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>fastjson</artifactId>
+            <version>1.2.51</version>
+        </dependency>
+        <dependency>
+            <groupId>cn.hutool</groupId>
+            <artifactId>hutool-all</artifactId>
+            <version>5.2.5</version>
+        </dependency>
         <dependency>
             <groupId>org.junit.jupiter</groupId>
             <artifactId>junit-jupiter</artifactId>
diff --git a/src/main/java/com/zcunsoft/App.java b/src/main/java/com/zcunsoft/App.java
index d45d7f664e730bb37b89b7c055c587f3be9d1a00..8a4d80e9c1fa54b79ad2b784ea3422a0f0a336a8 100644
--- a/src/main/java/com/zcunsoft/App.java
+++ b/src/main/java/com/zcunsoft/App.java
@@ -1,9 +1,10 @@
 package com.zcunsoft;
 
+import org.mybatis.spring.annotation.MapperScan;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 
-
+@MapperScan("com.zcunsoft.*.mapper")
 @SpringBootApplication
 public class App {
 
diff --git a/src/main/java/com/zcunsoft/cfg/KafkaConfig.java b/src/main/java/com/zcunsoft/cfg/KafkaConfig.java
index 1d5e45194dc23ec538aea11bd1842fe24903f4b1..1bdd4784b8a11d8400b871a5a93cf50474a37da3 100644
--- a/src/main/java/com/zcunsoft/cfg/KafkaConfig.java
+++ b/src/main/java/com/zcunsoft/cfg/KafkaConfig.java
@@ -1,12 +1,14 @@
 package com.zcunsoft.cfg;
 
 import org.apache.kafka.clients.admin.NewTopic;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
 import javax.annotation.Resource;
 
 @Configuration
+@ConditionalOnProperty(name = "log.model", havingValue = "kafka")
 public class KafkaConfig {
 
 	@Resource
diff --git a/src/main/java/com/zcunsoft/controller/KafkaController.java b/src/main/java/com/zcunsoft/controller/KafkaController.java
index 6538ced2ce9812414ba21b0ca52fd7e4e1eecdcf..bea22955a864130bb18b924425309e7f60328e0e 100644
--- a/src/main/java/com/zcunsoft/controller/KafkaController.java
+++ b/src/main/java/com/zcunsoft/controller/KafkaController.java
@@ -41,25 +41,33 @@ public class KafkaController {
     private ReceiverObjectMapper objectMapper;
 
     @Resource
-    private  ReceiverSetting serverSettings;
+    private ReceiverSetting serverSettings;
 
     @RequestMapping(value = "api/gp", method = {RequestMethod.GET, RequestMethod.POST})
     public ResponseEntity<Object> sendGet(QueryCriteria queryCriteria, HttpServletRequest request) {
-        if(serverSettings.getAppList().contains(queryCriteria.getProject())) {
+        // 如果serverSettings中的AppList包含queryCriteria中的project
+        if (serverSettings.getAppList().contains(queryCriteria.getProject())) {
+            // 获取请求体字符串
             String bodyString = getBodyString(request);
+            // 按照"&"字符分割请求体字符串为数组
             String[] bodyStringList = bodyString.split("&");
+            // 如果请求体只有一个参数且不为空
             if (bodyStringList.length == 1 && !bodyString.equals(""))
                 try {
+                    // 将请求体字符串转换为JsonNode对象
                     JsonNode jsonNode = objectMapper.readTree(bodyString);
-
+                    // 从JsonNode中提取并设置queryCriteria的gzip、data_list、data和crc属性
                     queryCriteria.setGzip(jsonNode.get("gzip").asText());
                     queryCriteria.setData_list(jsonNode.get("data_list").asText());
                     queryCriteria.setData(jsonNode.get("data").asText());
                     queryCriteria.setCrc(jsonNode.get("crc").asText());
                 } catch (Exception exception) {
                 }
+            // 遍历请求体参数列表
             for (String s : bodyStringList) {
+                // 按等号拆分参数名与值
                 String[] elm = s.split("=");
+                // 根据参数名设置queryCriteria的相关属性
                 if (elm[0].equals("data_list")) {
                     queryCriteria.setData_list(elm[1]);
                 } else if (elm[0].equals("data")) {
@@ -70,9 +78,12 @@ public class KafkaController {
                     queryCriteria.setGzip(elm[1]);
                 }
             }
+            // 获取请求的IP地址
             String ip = getIpAddr(request);
             try {
+                // 初始化变量用于存储最终的数据内容
                 String dataFinal, decodedString = null;
+                // 根据data_list或data的值进行解码操作
                 if (queryCriteria.getData_list() != null) {
                     if (Pattern.matches(".*\\+.*|.*\\/.*|.*=.*", queryCriteria.getData_list())) {
                         decodedString = queryCriteria.getData_list();
@@ -88,6 +99,7 @@ public class KafkaController {
                 } else {
                     logger.error("data为空");
                 }
+                // 如果decodedString不为空，则进一步进行Base64解码和可能的GZIP解压操作
                 if (decodedString != null) {
                     Base64.Decoder decoder = Base64.getDecoder();
                     byte[] byteArrayNEW = decoder.decode(decodedString);
@@ -97,17 +109,21 @@ public class KafkaController {
                     } else {
                         dataFinal = new String(byteArrayNEW);
                     }
+                    // 设置queryCriteria的data属性和客户端IP
                     queryCriteria.setData(dataFinal);
                     queryCriteria.setClientIp(ip);
+                    // 获取并设置user-agent信息
                     String ua = request.getHeader("user-agent");
                     if (ua == null) {
                         ua = request.getHeader("User-Agent");
                     }
                     queryCriteria.setUa(ua);
+                    // 将处理后的queryCriteria放入队列中
                     constsDataHolder.getLogQueue().put(queryCriteria);
                     logger.info(ip + "," + dataFinal);
                 }
             } catch (Exception e) {
+                // 当出现异常时，将queryCriteria的内容记录到日志，并打印异常堆栈
                 String logData = queryCriteria.toString();
                 logger.error(logData, e);
             }
diff --git a/src/main/java/com/zcunsoft/daemon/LogReceiveProcessBoss.java b/src/main/java/com/zcunsoft/daemon/LogReceiveProcessBoss.java
index b6b53661d8d01dd432ed954f4c1a248086a1fd5c..a633adc2444a3fc950cee16f7145b3b58ff57c8a 100644
--- a/src/main/java/com/zcunsoft/daemon/LogReceiveProcessBoss.java
+++ b/src/main/java/com/zcunsoft/daemon/LogReceiveProcessBoss.java
@@ -3,9 +3,12 @@ package com.zcunsoft.daemon;
 import com.zcunsoft.cfg.ReceiverSetting;
 import com.zcunsoft.dto.QueryCriteria;
 import com.zcunsoft.handlers.ConstsDataHolder;
+import com.zcunsoft.LogAnalysis.service.ILogAnalysisService;
 import com.zcunsoft.services.IReceiveService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Lazy;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.PostConstruct;
@@ -32,6 +35,13 @@ public class LogReceiveProcessBoss {
 
     boolean running = false;
 
+    @Resource()
+    @Lazy
+    private ILogAnalysisService iLogService;
+
+    @Value("${log.model:mp}")
+    private String model;
+
     @PostConstruct
     public void start() throws Exception {
         threadList = new ArrayList<Thread>();
@@ -91,7 +101,13 @@ public class LogReceiveProcessBoss {
     }
 
     private void handle(List<QueryCriteria> logList) {
-        ireceiveService.enqueueKafka(logList);
+        if (model.equals("kafka")) {
+            ireceiveService.enqueueKafka(logList);
+        } else {
+            iLogService.saveClickhouse(logList);
+        }
+
+
     }
 
 
diff --git a/src/main/java/com/zcunsoft/services/IReceiveService.java b/src/main/java/com/zcunsoft/services/IReceiveService.java
index f38886a84ef3bf369cb0634bab62b4d1bb16aec6..dc16907ab7f7d10292b6c8d09f3066da249bdc84 100644
--- a/src/main/java/com/zcunsoft/services/IReceiveService.java
+++ b/src/main/java/com/zcunsoft/services/IReceiveService.java
@@ -19,4 +19,6 @@ public interface IReceiveService {
     void loadCountry();
 
     void loadUrlRule();
+
+
 }
diff --git a/src/main/java/com/zcunsoft/services/ReceiveServiceImpl.java b/src/main/java/com/zcunsoft/services/ReceiveServiceImpl.java
index ff7ebb08fb226b6b2fd2a2c8675e56f58357ad6c..fb2d4e90ebe31f58db6a543696c7b9cc0f181c16 100644
--- a/src/main/java/com/zcunsoft/services/ReceiveServiceImpl.java
+++ b/src/main/java/com/zcunsoft/services/ReceiveServiceImpl.java
@@ -60,6 +60,7 @@ public class ReceiveServiceImpl implements IReceiveService {
     private final TypeReference<Rule> ruleTypeReference = new TypeReference<Rule>() {
     };
 
+
     public ReceiveServiceImpl(ConstsDataHolder constsDataHolder, ReceiverObjectMapper objectMapper, StringRedisTemplate queueRedisTemplate, AbstractUserAgentAnalyzer userAgentAnalyzer, KafkaTemplate<String, String> kafkaTemplate, ReceiverSetting serverSettings) {
         this.objectMapper = objectMapper;
         this.queueRedisTemplate = queueRedisTemplate;
@@ -208,6 +209,7 @@ public class ReceiveServiceImpl implements IReceiveService {
         }
     }
 
+
     private void setProperty(JsonNode jsonObject, String ua, Region region) {
         try {
             if (jsonObject.get("properties") != null) {
diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml
index 42a2039b087a1a4e198803dfb6e65e0c4d93b17f..da589cd0e2fa19b1ce9b87764d0a34f35c4e7198 100644
--- a/src/main/resources/application.yml
+++ b/src/main/resources/application.yml
@@ -1,5 +1,10 @@
 server:
   port: 8002
+  tomcat:
+    #日志配配置
+    accesslog:
+      #目录
+      directory: ./receiverlogs
 logging:
   file:
     path: log
@@ -31,11 +36,33 @@ spring:
     # 单机配置
     host: localhost
     port: 6379
-#    password: nW2zFwS41tdf
+    password:
      # 哨兵配置
 #    sentinel:
 #      master: gct
 #      nodes: 10.100.2.1:26379,10.100.2.2:26379,10.100.2.3:26379
+  datasource:
+    type: com.alibaba.druid.pool.DruidDataSource
+    url: jdbc:clickhouse://localhost:8123/clklog
+    username: default
+    password: 123456
+    driver-class-name: ru.yandex.clickhouse.ClickHouseDriver
+    initialSize: 10
+    maxActive: 100
+    minIdle: 10
+    maxWait: 6000
+
+mybatis:
+  mapper-locations: classpath:mapper/*Mapper.xml
+  configuration:
+    log-impl: org.apache.ibatis.logging.stdout.StdOutImpl
+    # 是否开启自动驼峰命名规则映射:从数据库列名到Java属性驼峰命名的类似映射
+    map-underscore-to-camel-case: true
+
 receiver:
   thread-count: 2
   app-list: clklogapp
+
+# mp:mybatisPlus
+log:
+  model: mp # kafka
diff --git a/src/main/resources/mapper/LogAnalysisMapper.xml b/src/main/resources/mapper/LogAnalysisMapper.xml
new file mode 100644
index 0000000000000000000000000000000000000000..c14ec1c2d9d0ddec6e6a96d21f40d98c8ff96a0a
--- /dev/null
+++ b/src/main/resources/mapper/LogAnalysisMapper.xml
@@ -0,0 +1,5 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="com.zcunsoft.LogAnalysis.mapper.LogAnalysisMapper">
+
+</mapper>