first commit

.gitignore

@@ -0,0 +1,5 @@
+.gradle
+.idea
+gradle
+gradlew
+gradlew.bat

Dockerfile

@@ -0,0 +1,8 @@
+FROM amazoncorretto:17-alpine-jdk
+
+RUN sed -i 's/dl-cdn.alpinelinux.org/mirrors.ustc.edu.cn/g' /etc/apk/repositories
+RUN apk update && apk add tzdata
+RUN ln -sf /usr/share/zoneinfo/Asia/Shanghai /etc/localtime
+RUN echo "Asia/Shanghai" > /etc/timezone
+COPY build/libs/app.jar /play.jar
+COPY conf/* /conf

ali-kafka.yml

@@ -0,0 +1,38 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: ali-kafka
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      k8s-app: ali-kafka
+  strategy:
+    type: Recreate
+  template:
+    metadata:
+      labels:
+        k8s-app: ali-kafka
+    spec:
+      imagePullSecrets:
+        - name: registry-key
+      containers:
+        - command: [ "java","-jar","/app.jar" ]
+          env:
+            - name: PATH
+              value: /usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
+            - name: TZ
+              value: Asia/Shanghai
+          image: { { image } }
+          imagePullPolicy: IfNotPresent
+          name: ali-kafka
+          resources:
+            limits:
+              cpu: 100m
+              memory: 515Mi
+            requests:
+              cpu: 100m
+              memory: 50Mi
+      dnsPolicy: ClusterFirst
+      terminationGracePeriodSeconds: 10
+

apply.sh

@@ -0,0 +1,7 @@
+
+version="1.0"
+docker build -t registry.cn-chengdu.aliyuncs.com/cxzx-spider/ali-kafka:$version .
+docker push registry.cn-chengdu.aliyuncs.com/cxzx-spider/ali-kafka:$version
+sed -i "s!{ { image } }!registry\.cn-chengdu\.aliyuncs\.com/cxzx-spider/ali-kafka:$version!g" *.yml
+
+kubectl apply -f ali-kafka.yml

build.gradle

@@ -0,0 +1,49 @@
+plugins {
+    id 'java'
+}
+
+group 'org.example'
+version '1.0-SNAPSHOT'
+repositories {
+    maven { url 'https://maven.aliyun.com/nexus/content/groups/public' }
+    mavenCentral()
+}
+
+dependencies {
+    implementation 'commons-io:commons-io:2.11.0'
+    implementation 'org.apache.commons:commons-lang3:3.12.0'
+    implementation 'com.fasterxml.jackson.core:jackson-databind:2.14.0'
+    implementation 'cn.hutool:hutool-all:5.8.9'
+
+    implementation 'ch.qos.logback:logback-classic:1.2.11'
+    implementation 'org.slf4j:slf4j-api:1.7.36'
+    implementation 'org.slf4j:log4j-over-slf4j:1.7.36'
+
+    implementation 'org.apache.kafka:kafka-clients:2.4.0'
+
+
+    implementation 'com.mysql:mysql-connector-j:8.0.31'
+
+    testImplementation 'org.junit.jupiter:junit-jupiter-api:5.8.1'
+    testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine:5.8.1'
+    compileOnly 'org.projectlombok:lombok:1.18.24'
+    annotationProcessor 'org.projectlombok:lombok:1.18.24'
+}
+
+jar {
+    manifest {
+        attributes(
+                "Manifest-Version": 1.0,
+                "Main-Class": "org.example.Main")
+    }
+
+    archiveFileName = 'app.jar'
+    from { configurations.compileClasspath.collect { it.isDirectory() ? it : zipTree(it) } }
+    into('assets') {
+        from 'assets'
+    }
+}
+
+test {
+    useJUnitPlatform()
+}

build.sh

@@ -0,0 +1,6 @@
+cd `dirname $0`
+git reset HEAD --hard
+git pull
+gradle build
+sh apply.sh
+

build/resources/main/logback.xml

@@ -0,0 +1,16 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration>
+    <property name="logBase" value="./logs"/>
+    <!--控制台输出日志信息-->
+    <appender name="CONSOLE-LOG" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} %-5level [%class:%line] - %msg%n</pattern>
+        </encoder>
+    </appender>
+
+
+
+    <root level="warn">
+        <appender-ref ref="CONSOLE-LOG"/>
+    </root>
+</configuration>

conf/kafka_client_jaas.conf

@@ -0,0 +1,5 @@
+KafkaClient {
+  org.apache.kafka.common.security.plain.PlainLoginModule required
+  username="alikafka_pre-cn-tl32wxejg003"
+  password="3N07SaSUQyT4ETxlR48f9mEwpWWQVN86";
+};

out/production/resources/logback.xml

@@ -0,0 +1,15 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration>
+    <!--控制台输出日志信息-->
+    <appender name="CONSOLE-LOG" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} %-5level [%class:%line] - %msg%n</pattern>
+        </encoder>
+    </appender>
+
+
+
+    <root level="warn">
+        <appender-ref ref="CONSOLE-LOG"/>
+    </root>
+</configuration>

settings.gradle

@@ -0,0 +1,2 @@
+rootProject.name = 'ali-kafka'
+

src/main/java/com/cxzx/Main.java

@@ -0,0 +1,74 @@
+package com.cxzx;
+
+import com.cxzx.db.Tidb;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.io.FilenameUtils;
+import org.apache.commons.lang3.SystemUtils;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.config.SaslConfigs;
+import org.apache.kafka.common.config.SslConfigs;
+import com.cxzx.db.KafkaData;
+
+import java.sql.Timestamp;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+@Slf4j
+public class Main {
+    public static void main(String[] args) {
+
+        var userDir = SystemUtils.getUserDir();
+        var jksPath = FilenameUtils.concat(userDir.getAbsolutePath(), "conf/only.4096.client.truststore.jks");
+        var jaasPath = FilenameUtils.concat(userDir.getAbsolutePath(), "conf/kafka_client_jaas.conf");
+        var props = new Properties();
+        //根证书store的密码，保持不变。
+        //接入协议，目前支持使用SASL_SSL协议接入。
+        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
+        props.put(CommonClientConfigs.GROUP_ID_CONFIG, "cxzx");
+        //SASL鉴权方式，保持不变。
+        props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
+        props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, jksPath);
+        //根证书store的密码，保持不变。
+        props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");
+        props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");
+        System.setProperty("java.security.auth.login.config", jaasPath);
+        props.put("bootstrap.servers",
+                "alikafka-pre-cn-tl32wxejg003-1.alikafka.aliyuncs.com:9093,alikafka-pre-cn-tl32wxejg003-2.alikafka.aliyuncs.com:9093,alikafka-pre-cn-tl32wxejg003-3.alikafka.aliyuncs.com:9093");
+        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
+        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
+        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
+
+        //注意该值不要改得太大，如果Poll太多数据，而不能在下次Poll之前消费完，则会触发一次负载均衡，产生卡顿。
+        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 300);
+        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 30 * 1000);
+
+
+        var consumer = new KafkaConsumer<String, String>(props);
+
+        try (consumer) {
+            consumer.subscribe(List.of("article"));
+            while (true) {
+                var kafkaDataList = new ArrayList<KafkaData>();
+                for (ConsumerRecord<String, String> record : consumer.poll(Duration.ofMillis(30 * 1000))) {
+                    kafkaDataList.add(new KafkaData(record.offset(), record.value(), new Timestamp(record.timestamp())));
+                    var offsetMap = Map.of(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset()));
+                    consumer.commitSync(offsetMap);
+                }
+                Tidb.insert(kafkaDataList);
+            }
+        } catch (Exception e) {
+            log.error(e.getMessage(), e);
+        }
+
+
+    }
+}

src/main/java/com/cxzx/db/KafkaData.java

@@ -0,0 +1,14 @@
+package com.cxzx.db;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+import java.sql.Timestamp;
+
+@Data
+@AllArgsConstructor
+public class KafkaData {
+    private Long offset;
+    private String data;
+    private Timestamp createTime;
+}

src/main/java/com/cxzx/db/Tidb.java

@@ -0,0 +1,43 @@
+package com.cxzx.db;
+
+import cn.hutool.core.collection.CollUtil;
+import lombok.Cleanup;
+import lombok.SneakyThrows;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.util.List;
+
+public class Tidb {
+
+    private static final Connection connection ;
+
+    static {
+        try {
+            Class.forName("com.mysql.cj.jdbc.Driver");
+            connection = DriverManager.getConnection("jdbc:mysql://172.16.101.1:4000/huaqiyun?autoReconnect=true&useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=CONVERT_TO_NULL&useSSL=false&nullCatalogMeansCurrent=true&rewriteBatchedStatements=true",
+                    "root", "msstar");
+
+        } catch (ClassNotFoundException | SQLException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    @SneakyThrows
+    public static void insert(List<KafkaData> kafkaDataList) {
+        if (CollUtil.isEmpty(kafkaDataList)) {
+            return;
+        }
+        @Cleanup
+        var preparedStatement = connection.prepareStatement("insert ignore into article (offset,create_time,data) values (?,?,?)");
+        for (KafkaData kafkaData : kafkaDataList) {
+            preparedStatement.setLong(1, kafkaData.getOffset());
+            preparedStatement.setTimestamp(2, kafkaData.getCreateTime());
+            preparedStatement.setString(3, kafkaData.getData());
+            preparedStatement.addBatch();
+        }
+        preparedStatement.executeBatch();
+
+    }
+}

src/main/resources/logback.xml

@@ -0,0 +1,15 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration>
+    <!--控制台输出日志信息-->
+    <appender name="CONSOLE-LOG" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} %-5level [%class:%line] - %msg%n</pattern>
+        </encoder>
+    </appender>
+
+
+
+    <root level="warn">
+        <appender-ref ref="CONSOLE-LOG"/>
+    </root>
+</configuration>