six-vehicle
/
cloud-server
5
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(): 新增kafka配置模块

dev.eventProcess
xinzirun 2024-09-28 21:23:24 +08:00
parent 49d6651df2
commit 38a096ea72
7 changed files with 303 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
cloud-common/cloud-common-kafka/pom.xml 100644
View File

@ -0,0 +1,33 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.muyu</groupId>
<artifactId>cloud-common</artifactId>
<version>3.6.3</version>
</parent>
<artifactId>cloud-common-kafka</artifactId>
<properties>
<maven.compiler.source>17</maven.compiler.source>
<maven.compiler.target>17</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<!-- 项目公共核心模块 -->
<dependency>
<groupId>com.muyu</groupId>
<artifactId>cloud-common-core</artifactId>
</dependency>
<!-- kafka客户端 -->
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
</dependency>
</dependencies>
</project>

102
cloud-common/cloud-common-kafka/src/main/java/com/muyu/common/kafka/config/KafkaConsumerConfig.java 100644
View File

@ -0,0 +1,102 @@
package com.muyu.common.kafka.config;
import com.muyu.common.core.text.StrFormatter;
import com.muyu.common.kafka.constant.KafkaConfigConstants;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
/**
* @Author: zi run
* @Date 2024/9/28 20:32
* @Description Kafka消费者配置
*/
@Configuration
public class KafkaConsumerConfig {
/**
* 服务端IP
*/
@Value("${kafka.consumer.bootstrap-servers-ip}")
private String bootstrapServersIP;
/**
* 服务端口号
*/
@Value("${kafka.consumer.bootstrap-servers-port}")
private String bootstrapServersPort;
/**
* 开启消费者偏移量
*/
@Value("${kafka.consumer.enable-auto-commit}")
private Boolean enableAutoCommit;
/**
* 自动提交时间间隔
*/
@Value("${kafka.consumer.auto-commit-interval}")
private Integer autoCommitInterval;
/**
* 自动重置偏移量
*/
@Value("${kafka.consumer.auto-offset-reset}")
private String autoOffsetReset;
/**
* 请求阻塞的最大时间
*/
@Value("${kafka.consumer.fetch-max-wait}")
private Integer fetchMaxWait;
/**
* 请求应答的最小字节数
*/
@Value("${kafka.consumer.fetch-min-size}")
private Integer fetchMinSize;
/**
* 心跳间隔时间
*/
@Value("${kafka.consumer.heartbeat-interval}")
private Integer heartbeatInterval;
/**
* 一次调用poll返回的最大记录条数
*/
@Value("${kafka.consumer.max-poll-records}")
private Integer maxPollRecords;
/**
* 指定消费组
*/
@Value("${kafka.consumer.group-id}")
private String groupId;
/**
* Kafka消费者初始化配置
* @return Kafka消费者实例
*/
@Bean
public KafkaConsumer<String, String> kafkaConsumer() {
HashMap<String, Object> configs = new HashMap<>();
configs.put(KafkaConfigConstants.BOOTSTRAP_SERVERS,
StrFormatter.format("{}:{}", bootstrapServersIP, bootstrapServersPort));
configs.put(KafkaConfigConstants.ENABLE_AUTO_COMMIT, enableAutoCommit);
configs.put(KafkaConfigConstants.AUTO_COMMIT_INTERVAL, autoCommitInterval);
configs.put(KafkaConfigConstants.AUTO_OFFSET_RESET, autoOffsetReset);
configs.put(KafkaConfigConstants.FETCH_MAX_WAIT, fetchMaxWait);
configs.put(KafkaConfigConstants.FETCH_MIN_SIZE, fetchMinSize);
configs.put(KafkaConfigConstants.HEARTBEAT_INTERVAL, heartbeatInterval);
configs.put(KafkaConfigConstants.MAX_POLL_RECORDS, maxPollRecords);
configs.put(KafkaConfigConstants.GROUP_ID, groupId);
Deserializer<String> keyDeserializer = new StringDeserializer();
Deserializer<String> valueDeserializer = new StringDeserializer();
return new KafkaConsumer<>(configs, keyDeserializer, valueDeserializer);
}
}

74
cloud-common/cloud-common-kafka/src/main/java/com/muyu/common/kafka/config/KafkaProducerConfig.java 100644
View File

@ -0,0 +1,74 @@
package com.muyu.common.kafka.config;
import com.muyu.common.core.text.StrFormatter;
import com.muyu.common.kafka.constant.KafkaConfigConstants;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
/**
* @Author: zi run
* @Date 2024/9/28 16:35
* @Description Kafka生产者配置
*/
@Configuration
public class KafkaProducerConfig {
/**
* 服务端IP
*/
@Value("${kafka.producer.bootstrap-servers-ip}")
private String bootstrapServersIP;
/**
* 服务端口号
*/
@Value("${kafka.producer.bootstrap-servers-port}")
private String bootstrapServersPort;
/**
* 重试次数
*/
@Value("${kafka.producer.retries}")
private Integer retries;
/**
* 默认批量大小
*/
@Value("${kafka.producer.batch-size}")
private Integer batchSize;
/**
* 总内存字节数
*/
@Value("${kafka.producer.buffer-memory}")
private Integer bufferMemory;
/**
* 偏移量
*/
@Value("${kafka.producer.acks}")
private String acks;
/**
* Kafka生产者初始化配置
* @return kafka生产者实例
*/
@Bean
public KafkaProducer<String, String> kafkaProducer() {
HashMap<String, Object> configs = new HashMap<>();
configs.put(KafkaConfigConstants.BOOTSTRAP_SERVERS,
StrFormatter.format("{}:{}", bootstrapServersIP, bootstrapServersPort));
configs.put(KafkaConfigConstants.RETRIES, retries);
configs.put(KafkaConfigConstants.BATCH_SIZE, batchSize);
configs.put(KafkaConfigConstants.BUFFER_MEMORY, bufferMemory);
configs.put(KafkaConfigConstants.ACKS, acks);
Serializer<String> keySerializer = new StringSerializer();
Serializer<String> valueSerializer = new StringSerializer();
return new KafkaProducer<>(configs, keySerializer, valueSerializer);
}
}

74
cloud-common/cloud-common-kafka/src/main/java/com/muyu/common/kafka/constant/KafkaConfigConstants.java 100644
View File

@ -0,0 +1,74 @@
package com.muyu.common.kafka.constant;
/**
* @Author: zi run
* @Date 2024/9/28 20:07
* @Description Kafka配置通用常量
*/
public class KafkaConfigConstants {
/**
* 服务端ip+端口号
*/
public static final String BOOTSTRAP_SERVERS = "bootstrap.servers";
/**
* 重试次数
*/
public static final String RETRIES = "retries";
/**
* 默认批量大小
*/
public static final String BATCH_SIZE = "batch.size";
/**
* 总内存字节数
*/
public static final String BUFFER_MEMORY = "buffer-memory";
/**
* 偏移量
*/
public static final String ACKS = "acks";
/**
* 开启消费者偏移量
*/
public static final String ENABLE_AUTO_COMMIT = "enable.auto.commit";
/**
* 自动提交时间间隔
*/
public static final String AUTO_COMMIT_INTERVAL = "auto.commit.interval";
/**
* 自动重置偏移量
*/
public static final String AUTO_OFFSET_RESET = "auto.offset.reset";
/**
* 请求阻塞的最大时间
*/
public static final String FETCH_MAX_WAIT = "fetch.max.wait";
/**
* 请求应答的最小字节数
*/
public static final String FETCH_MIN_SIZE = "fetch.min.size";
/**
* 心跳间隔时间
*/
public static final String HEARTBEAT_INTERVAL = "heartbeat-interval";
/**
* 一次调用poll返回的最大记录条数
*/
public static final String MAX_POLL_RECORDS = "max.poll.records";
/**
* 指定消费组
*/
public static final String GROUP_ID = "group.id";
}

2
cloud-common/cloud-common-kafka/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports 100644
View File

@ -0,0 +1,2 @@
com.muyu.common.kafka.config.KafkaProducerConfig
com.muyu.common.kafka.config.KafkaConsumerConfig

1
cloud-common/pom.xml
View File

@ -21,6 +21,7 @@
<module>cloud-common-system</module>
<module>cloud-common-xxl</module>
<module>cloud-common-rabbit</module>
<module>cloud-common-kafka</module>
</modules>
<artifactId>cloud-common</artifactId>

18
pom.xml
View File

@ -43,6 +43,7 @@
<knife4j-openapi3.version>4.1.0</knife4j-openapi3.version>
<xxl-job-core.version>2.4.1</xxl-job-core.version>
<swagger.an.jakarta.verison>2.2.8</swagger.an.jakarta.verison>
<kafka.clients.verison>3.0.0</kafka.clients.verison>
</properties>
<!-- 依赖声明 -->
@ -184,12 +185,20 @@
<version>${transmittable-thread-local.version}</version>
</dependency>
<!-- swagger3文档 -->
<dependency>
<groupId>io.swagger.core.v3</groupId>
<artifactId>swagger-annotations-jakarta</artifactId>
<version>${swagger.an.jakarta.verison}</version>
</dependency>
<!-- Kafka客户端 -->
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>${kafka.clients.verison}</version>
</dependency>
<!-- 核心模块 -->
<dependency>
<groupId>com.muyu</groupId>
@ -281,7 +290,14 @@
<version>${muyu.version}</version>
</dependency>
<!-- 企业业务平台 - 公告模块 -->
<!-- kafka模块 -->
<dependency>
<groupId>com.muyu</groupId>
<artifactId>cloud-common-kafka</artifactId>
<version>${muyu.version}</version>
</dependency>
<!-- 企业业务平台 - 公共模块 -->
<dependency>
<groupId>com.muyu</groupId>
<artifactId>cloud-modules-enterprise-common</artifactId>