Shikong/springcloud-study
1
0
Fork 0
mirror of https://gitee.com/shikong-sk/springcloud-study synced 2024-11-09 01:51:19 +08:00
Code Issues Packages Projects Releases Wiki Activity

kafka 简单试水

Browse Source
This commit is contained in:
Shikong 2023-04-21 17:01:17 +08:00
parent caa2668eef
commit 8ae54c0e1c
15 changed files with 334 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
.idea/compiler.xml
View File

@ -7,12 +7,17 @@
<sourceOutputDir name="target/generated-sources/annotations" />
<sourceTestOutputDir name="target/generated-test-sources/test-annotations" />
<outputRelativeToContentRoot value="true" />
<module name="kafka-01-producer" />
<module name="kafka-01-common" />
<module name="provider" />
<module name="kafka-01" />
<module name="consumer" />
<module name="kafka-01-consumer" />
</profile>
</annotationProcessing>
<bytecodeTargetLevel>
<module name="consumer (1) (cn.skcks.study.springcloud.kafka.consumer)" target="1.8" />
<module name="consumer (2) (cn.skcks.study.springcloud.project01)" target="1.8" />
<module name="producer" target="1.8" />
<module name="springcloud-study" target="1.5" />
</bytecodeTargetLevel>
</component>

6
.idea/encodings.xml
View File

@ -1,6 +1,12 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="Encoding">
<file url="file://$PROJECT_DIR$/kafka-01/kafka-01-common/src/main/java" charset="UTF-8" />
<file url="file://$PROJECT_DIR$/kafka-01/kafka-01-common/src/main/resources" charset="UTF-8" />
<file url="file://$PROJECT_DIR$/kafka-01/kafka-01-consumer/src/main/java" charset="UTF-8" />
<file url="file://$PROJECT_DIR$/kafka-01/kafka-01-consumer/src/main/resources" charset="UTF-8" />
<file url="file://$PROJECT_DIR$/kafka-01/kafka-01-producer/src/main/java" charset="UTF-8" />
<file url="file://$PROJECT_DIR$/kafka-01/kafka-01-producer/src/main/resources" charset="UTF-8" />
<file url="file://$PROJECT_DIR$/kafka-01/src/main/java" charset="UTF-8" />
<file url="file://$PROJECT_DIR$/kafka-01/src/main/resources" charset="UTF-8" />
<file url="file://$PROJECT_DIR$/project-01/consumer/src/main/java" charset="UTF-8" />

21
kafka-01/kafka-01-common/pom.xml Normal file
View File

@ -0,0 +1,21 @@
<?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>cn.skcks.study.springcloud.kafka</groupId>
<artifactId>kafka-01</artifactId>
<version>1.0-SNAPSHOT</version>
</parent>
<groupId>cn.skcks.study.springcloud.kafka.common</groupId>
<artifactId>kafka-01-common</artifactId>
<properties>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
</project>

8
kafka-01/kafka-01-common/src/main/java/cn/skcks/study/springcloud/kafka/common/model/dto/Message.java Normal file
View File

@ -0,0 +1,8 @@
package cn.skcks.study.springcloud.kafka.common.model.dto;
import lombok.Data;
@Data
public class Message {
private String data;
}

81
kafka-01/kafka-01-common/src/main/java/cn/skcks/study/springcloud/kafka/common/utils/JsonUtils.java Normal file
View File

@ -0,0 +1,81 @@
package cn.skcks.study.springcloud.kafka.common.utils;
import com.fasterxml.jackson.core.json.JsonReadFeature;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
@SuppressWarnings("unused")
public class JsonUtils {
public static final ObjectMapper mapper;
public static final ObjectMapper compressMapper;
static {
mapper = new ObjectMapper();
compressMapper = new ObjectMapper();
mapper.enable(SerializationFeature.INDENT_OUTPUT);
compressMapper.disable(SerializationFeature.INDENT_OUTPUT);
// 返回内容中 不含 值为 null 的字段
// mapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
mapper.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false);
compressMapper.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false);
// 如果json中有新增的字段并且是实体类类中不存在的，不报错
mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
mapper.configure(DeserializationFeature.FAIL_ON_IGNORED_PROPERTIES, false);
compressMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
compressMapper.configure(DeserializationFeature.FAIL_ON_IGNORED_PROPERTIES, false);
// 允许出现特殊字符和转义符
mapper.configure(JsonReadFeature.ALLOW_UNESCAPED_CONTROL_CHARS.mappedFeature(), true);
compressMapper.configure(JsonReadFeature.ALLOW_UNESCAPED_CONTROL_CHARS.mappedFeature(), true);
// 允许出现单引号
mapper.configure(JsonReadFeature.ALLOW_SINGLE_QUOTES.mappedFeature(), true);
compressMapper.configure(JsonReadFeature.ALLOW_SINGLE_QUOTES.mappedFeature(), true);
}
public static <T> T parse(String json, Class<T> clazz) {
try {
return mapper.readValue(json, clazz);
} catch (Exception e) {
e.printStackTrace();
return null;
}
}
public static <T> T parse(String json, TypeReference<T> clazz) {
try {
return mapper.readValue(json, clazz);
} catch (Exception e) {
e.printStackTrace();
return null;
}
}
public static String toCompressJson(Object obj) {
try {
return compressMapper.writeValueAsString(obj);
} catch (Exception e) {
e.printStackTrace();
return null;
}
}
public static String toJson(Object obj) {
try {
return mapper.writeValueAsString(obj);
} catch (Exception e) {
e.printStackTrace();
return null;
}
}
public static <T> T convert(Object object, Class<T> clazz) {
return JsonUtils.parse(JsonUtils.toJson(object), clazz);
}
}

28
kafka-01/kafka-01-consumer/pom.xml Normal file
View File

@ -0,0 +1,28 @@
<?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>cn.skcks.study.springcloud.kafka</groupId>
<artifactId>kafka-01</artifactId>
<version>1.0-SNAPSHOT</version>
</parent>
<groupId>cn.skcks.study.springcloud.kafka.consumer</groupId>
<artifactId>kafka-01-consumer</artifactId>
<properties>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>cn.skcks.study.springcloud.kafka.common</groupId>
<artifactId>kafka-01-common</artifactId>
<version>${project.version}</version>
</dependency>
</dependencies>
</project>

13
kafka-01/kafka-01-consumer/src/main/java/cn/skcks/study/springcloud/kafka/consumer/Kafka01Consumer.java Normal file
View File

@ -0,0 +1,13 @@
package cn.skcks.study.springcloud.kafka.consumer;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class Kafka01Consumer {
public static void main(String[] args) {
SpringApplication.run(Kafka01Consumer.class,args);
}
}

20
kafka-01/kafka-01-consumer/src/main/java/cn/skcks/study/springcloud/kafka/consumer/services/MQListener.java Normal file
View File

@ -0,0 +1,20 @@
package cn.skcks.study.springcloud.kafka.consumer.services;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
@Slf4j
@Component
public class MQListener {
@KafkaListener(topics = {"test"},groupId = "test-group")
public void onMessage(ConsumerRecord<String, String> record, Acknowledgment ack) {
log.info("消费消息: 主题 {} 分片 {} ",record.topic(),record.partition());
log.info("key {}, value {}",record.key(),record.value());
ack.acknowledge();
}
}

27
kafka-01/kafka-01-consumer/src/main/resources/application.yml Normal file
View File

@ -0,0 +1,27 @@
spring:
kafka:
# bootstrap-servers: 10.10.10.200:9092
bootstrap-servers: 10.10.10.200:9192,10.10.10.200:9292,10.10.10.200:9392
consumer:
# 自动提交的时间间隔 在spring boot 2.X 版本是值的类型为Duration 需要符合特定的格式，如1S,1M,2H,5D
auto-commit-interval: 1S
# 该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下该作何处理：
auto-offset-reset: earliest
# 是否自动提交偏移量，默认值是true,为了避免出现重复数据和数据丢失，可以把它设置为false,然后手动提交偏移量
enable-auto-commit: false
# 键的反序列化方式
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
# 值的反序列化方式
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
listener:
#手工ack，调用ack后立刻提交offset
ack-mode: manual_immediate
#容器运行的线程数
server:
port: 8082

38
kafka-01/kafka-01-producer/pom.xml Normal file
View File

@ -0,0 +1,38 @@
<?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>cn.skcks.study.springcloud.kafka</groupId>
<artifactId>kafka-01</artifactId>
<version>1.0-SNAPSHOT</version>
</parent>
<artifactId>kafka-01-producer</artifactId>
<properties>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>cn.skcks.study.springcloud.kafka.common</groupId>
<artifactId>kafka-01-common</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>2.9.8</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
</dependencies>
</project>

12
kafka-01/kafka-01-producer/src/main/java/cn/skcks/study/springcloud/kafka/producer/Kafka01Producer.java Normal file
View File

@ -0,0 +1,12 @@
package cn.skcks.study.springcloud.kafka.producer;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class Kafka01Producer {
public static void main(String[] args) {
SpringApplication.run(Kafka01Producer.class,args);
}
}

31
kafka-01/kafka-01-producer/src/main/java/cn/skcks/study/springcloud/kafka/producer/controller/SimpleController.java Normal file
View File

@ -0,0 +1,31 @@
package cn.skcks.study.springcloud.kafka.producer.controller;
import cn.skcks.study.springcloud.kafka.common.model.dto.Message;
import cn.skcks.study.springcloud.kafka.common.utils.JsonUtils;
import com.fasterxml.jackson.core.JsonProcessingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class SimpleController {
@Autowired
private KafkaTemplate<Object, String> kafkaTemplate;
final static String TOPIC_NAME = "test";
@GetMapping("/api/v2/{num}")
public void sendMessage1(@PathVariable("num") String num) throws JsonProcessingException {
Message message = new Message();
message.setData("这是一个消息,num="+num);
kafkaTemplate.send(TOPIC_NAME,num, JsonUtils.toJson(message)).addCallback(success->{
System.out.println("发送成功");
},failure->{
System.out.println("发送消息失败："+failure.getMessage());
});
}
}

19
kafka-01/kafka-01-producer/src/main/resources/application.yml Normal file
View File

@ -0,0 +1,19 @@
spring:
kafka:
# bootstrap-servers: 10.10.10.200:9092
bootstrap-servers: 10.10.10.200:9192,10.10.10.200:9292,10.10.10.200:9392
producer:
retries: 1
batch-size: 1024
buffer-memory: 10240
# 键的序列化方式
key-serializer: org.apache.kafka.common.serialization.StringSerializer
# 值的序列化方式
value-serializer: org.apache.kafka.common.serialization.StringSerializer
#配置事务的话：如果用户显式地指定了 acks 参数，那么这个参数的值必须-1 all
acks: all
server:
port: 8081

24
kafka-01/pom.xml
View File

@ -11,6 +11,12 @@
<groupId>cn.skcks.study.springcloud.kafka</groupId>
<artifactId>kafka-01</artifactId>
<packaging>pom</packaging>
<modules>
<module>kafka-01-consumer</module>
<module>kafka-01-producer</module>
<module>kafka-01-common</module>
</modules>
<properties>
<maven.compiler.source>8</maven.compiler.source>
@ -18,4 +24,22 @@
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>2.9.8</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<scope>provided</scope>
</dependency>
</dependencies>
</project>

7
kafka-01/src/main/java/cn/skcks/study/springcloud/kafka/Main.java
View File

@ -1,7 +0,0 @@
package cn.skcks.study.springcloud.kafka;
public class Main {
public static void main(String[] args) {
System.out.println("Hello world!");
}
}