blog/spring-samples-for-all
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

增加spring boot kafka 用例

Browse Source
This commit is contained in:
罗祥
2019-01-08 17:51:31 +08:00
parent 72cafd9bbb
commit fc0c2b90b9
3 changed files with 24 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-boot/spring-boot-kafka/src/main/java/com/heibaiying/springboot/consumer/KafkaBeanConsumer.java
View File

@ -17,7 +17,7 @@ import org.springframework.stereotype.Component;
@Slf4j @Slf4j
public class KafkaBeanConsumer { public class KafkaBeanConsumer {
@KafkaListener(groupId = "group", topics = Topic.BEAN) @KafkaListener(groupId = "beanGroup", topics = Topic.BEAN)
public void consumer(ConsumerRecord<String, Object> record) { public void consumer(ConsumerRecord<String, Object> record) {
System.out.println("消费者收到消息:" + JSON.parseObject(record.value().toString(), Programmer.class)); System.out.println("消费者收到消息:" + JSON.parseObject(record.value().toString(), Programmer.class));
} }

6
spring-boot/spring-boot-kafka/src/main/java/com/heibaiying/springboot/consumer/KafkaSimpleConsumer.java
View File

@ -2,6 +2,7 @@ package com.heibaiying.springboot.consumer;
import com.heibaiying.springboot.constant.Topic; import com.heibaiying.springboot.constant.Topic;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener; import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders; import org.springframework.kafka.support.KafkaHeaders;
@ -18,8 +19,9 @@ import org.springframework.stereotype.Component;
public class KafkaSimpleConsumer { public class KafkaSimpleConsumer {
// 简单消费者 // 简单消费者
@KafkaListener(groupId = "group",topics = Topic.SIMPLE) @KafkaListener(groupId = "simpleGroup", topics = Topic.SIMPLE)
public void consumer1_1(ConsumerRecord<String, Object> record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) { public void consumer1_1(ConsumerRecord<String, Object> record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,Consumer consumer) {
System.out.println("消费者收到消息:" + record.value() + "; topic:" + topic); System.out.println("消费者收到消息:" + record.value() + "; topic:" + topic);
consumer.commitSync();
} }
} }

39
spring-boot/spring-boot-kafka/src/main/resources/application.yml
View File

@ -1,36 +1,35 @@
spring: spring:
kafka: kafka:
# <EFBFBD>Զ<EFBFBD><EFBFBD>ŷָ<EFBFBD><EFBFBD>ĵ<EFBFBD>ַ<EFBFBD>б<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ڽ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Kafka<EFBFBD><EFBFBD>Ⱥ<EFBFBD>ij<EFBFBD>ʼ<EFBFBD><EFBFBD><EFBFBD><EFBFBD>(kafka Ĭ<>ϵĶ˿ں<CBBF>Ϊ9092) # 以逗号分隔的地址列表，用于建立与Kafka集群的初始连接(kafka 默认的端口号为9092)
bootstrap-servers: 127.0.0.1:9092 bootstrap-servers: 127.0.0.1:9092
producer: producer:
# <EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ϣ<EFBFBD>ط<EFBFBD><EFBFBD>Ĵ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD> # 发生错误后，消息重发的次数。
retries: 0 retries: 0
#<EFBFBD><EFBFBD><EFBFBD>ж<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ϣ<EFBFBD><EFBFBD>Ҫ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>͵<EFBFBD>ͬһ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ʱ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>߻<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ƿ<EFBFBD><EFBFBD><EFBFBD>ͬһ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ò<EFBFBD><EFBFBD><EFBFBD>ָ<EFBFBD><EFBFBD><EFBFBD><EFBFBD>һ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ο<EFBFBD><EFBFBD><EFBFBD>ʹ<EFBFBD>õ<EFBFBD><EFBFBD>ڴ<EFBFBD><EFBFBD><EFBFBD>С<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ֽ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>㡣 #当有多个消息需要被发送到同一个分区时，生产者会把它们放在同一个批次里。该参数指定了一个批次可以使用的内存大小，按照字节数计算。
batch-size: 16384 batch-size: 16384
# <EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ڴ滺<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ĵ<EFBFBD>С<EFBFBD><EFBFBD> # 设置生产者内存缓冲区的大小。
buffer-memory: 33554432 buffer-memory: 33554432
# <EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>л<EFBFBD><EFBFBD><EFBFBD>ʽ # 键的序列化方式
key-serializer: org.apache.kafka.common.serialization.StringSerializer key-serializer: org.apache.kafka.common.serialization.StringSerializer
# ֵ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>л<EFBFBD><EFBFBD><EFBFBD>ʽ # 值的序列化方式
value-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.apache.kafka.common.serialization.StringSerializer
# acks=0 <EFBFBD><EFBFBD> <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ڳɹ<DAB3>д<EFBFBD><D0B4><EFBFBD><EFBFBD>Ϣ֮ǰ<D6AE><C7B0><EFBFBD><EFBFBD><EFBFBD>ȴ<EFBFBD><C8B4>κ<EFBFBD><CEBA><EFBFBD><EFBFBD>Է<EFBFBD><D4B7><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ӧ<EFBFBD><D3A6> # acks=0 ： 生产者在成功写入消息之前不会等待任何来自服务器的响应。
# acks=1 <EFBFBD><EFBFBD> ֻҪ<D6BB><D2AA>Ⱥ<EFBFBD><C8BA><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ڵ<EFBFBD><DAB5>յ<EFBFBD><D5B5><EFBFBD>Ϣ<EFBFBD><CFA2><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>߾ͻ<DFBE><CDBB>յ<EFBFBD>һ<EFBFBD><D2BB><EFBFBD><EFBFBD><EFBFBD>Է<EFBFBD><D4B7><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ɹ<EFBFBD><C9B9><EFBFBD>Ӧ<EFBFBD><D3A6> # acks=1 ： 只要集群的首领节点收到消息，生产者就会收到一个来自服务器成功响应。
# acks=all <EFBFBD><EFBFBD>ֻ<EFBFBD>е<EFBFBD><EFBFBD><EFBFBD><EFBFBD>в<EFBFBD><EFBFBD>븴<EFBFBD>ƵĽڵ<EFBFBD>ȫ<EFBFBD><EFBFBD><EFBFBD>յ<EFBFBD><EFBFBD><EFBFBD>Ϣʱ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>߲Ż<EFBFBD><EFBFBD>յ<EFBFBD>һ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Է<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ijɹ<EFBFBD><EFBFBD><EFBFBD>Ӧ<EFBFBD><EFBFBD> # acks=all ：只有当所有参与复制的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应。
acks: all acks: 1
consumer: consumer:
# <EFBFBD>Զ<EFBFBD><EFBFBD>ύ<EFBFBD><EFBFBD>ʱ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD> <20><>spring boot 2.X <20>汾<EFBFBD><E6B1BE><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>õ<EFBFBD><C3B5><EFBFBD>ֵ<EFBFBD><D6B5><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ΪDuration <20><>Ҫ<EFBFBD><D2AA><EFBFBD><EFBFBD><EFBFBD>ض<EFBFBD><D8B6>ĸ<EFBFBD>ʽ<EFBFBD><CABD><EFBFBD><EFBFBD>1S,1M,2H,5D # 自动提交的时间间隔 在spring boot 2.X 版本中这里采用的是值的类型为Duration 需要符合特定的格式，如1S,1M,2H,5D
auto-commit-interval: 1S auto-commit-interval: 1S
# <EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ָ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ڶ<EFBFBD>ȡһ<EFBFBD><EFBFBD>û<EFBFBD><EFBFBD>ƫ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ķ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ƫ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ч<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>¸<EFBFBD><EFBFBD><EFBFBD><EFBFBD>δ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD> # 该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下该作何处理：
# latest<EFBFBD><EFBFBD>Ĭ<EFBFBD><EFBFBD>ֵ<EFBFBD><EFBFBD><EFBFBD><EFBFBD>ƫ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ч<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>£<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>߽<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>µļ<EFBFBD>¼<EFBFBD><EFBFBD>ʼ<EFBFBD><EFBFBD>ȡ<EFBFBD><EFBFBD><EFBFBD>ݣ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>֮<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ɵļ<EFBFBD>¼<EFBFBD><EFBFBD> # latest（默认值）在偏移量无效的情况下，消费者将从最新的记录开始读取数据（在消费者启动之后生成的记录）
# earliest <EFBFBD><EFBFBD><EFBFBD><EFBFBD>ƫ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ч<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>£<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>߽<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ʼλ<EFBFBD>ö<EFBFBD>ȡ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ļ<EFBFBD>¼ # earliest ：在偏移量无效的情况下，消费者将从起始位置读取分区的记录
auto-offset-reset: earliest auto-offset-reset: earliest
# <EFBFBD>Ƿ<EFBFBD><EFBFBD>Զ<EFBFBD><EFBFBD>ύƫ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ĭ<EFBFBD><EFBFBD>ֵ<EFBFBD><EFBFBD>true,Ϊ<>˱<EFBFBD><CBB1><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ظ<EFBFBD><D8B8><EFBFBD><EFBFBD>ݺ<EFBFBD><DDBA><EFBFBD><EFBFBD>ݶ<EFBFBD>ʧ<EFBFBD><CAA7><EFBFBD><EFBFBD><EFBFBD>԰<EFBFBD><D4B0><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ϊfalse,Ȼ<><C8BB><EFBFBD>ֶ<EFBFBD><D6B6>ύƫ<E1BDBB><C6AB><EFBFBD><EFBFBD> # 是否自动提交偏移量，默认值是true,为了避免出现重复数据和数据丢失，可以把它设置为false,然后手动提交偏移量
enable-auto-commit: true enable-auto-commit: false
# <EFBFBD><EFBFBD><EFBFBD>ķ<EFBFBD><EFBFBD><EFBFBD><EFBFBD>л<EFBFBD><EFBFBD><EFBFBD>ʽ # 键的反序列化方式
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
# ֵ<EFBFBD>ķ<EFBFBD><EFBFBD><EFBFBD><EFBFBD>л<EFBFBD><EFBFBD><EFBFBD>ʽ # 值的反序列化方式
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
listener: listener:
# <EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>е<EFBFBD><EFBFBD>߳<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD> # 在侦听器容器中运行的线程数。
concurrency: 5 concurrency: 5