From fc0c2b90b9a2b69faf492657b10c3d191b3eb1f9 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E7=BD=97=E7=A5=A5?= <1366971433@qq.com>
Date: Tue, 8 Jan 2019 17:51:31 +0800
Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0spring=20boot=20=20kafka=20?=
 =?UTF-8?q?=E7=94=A8=E4=BE=8B?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../consumer/KafkaBeanConsumer.java           |  2 +-
 .../consumer/KafkaSimpleConsumer.java         |  6 ++-
 .../src/main/resources/application.yml        | 39 +++++++++----------
 3 files changed, 24 insertions(+), 23 deletions(-)

diff --git a/spring-boot/spring-boot-kafka/src/main/java/com/heibaiying/springboot/consumer/KafkaBeanConsumer.java b/spring-boot/spring-boot-kafka/src/main/java/com/heibaiying/springboot/consumer/KafkaBeanConsumer.java
index 540c30d..4bef598 100644
--- a/spring-boot/spring-boot-kafka/src/main/java/com/heibaiying/springboot/consumer/KafkaBeanConsumer.java
+++ b/spring-boot/spring-boot-kafka/src/main/java/com/heibaiying/springboot/consumer/KafkaBeanConsumer.java
@@ -17,7 +17,7 @@ import org.springframework.stereotype.Component;
 @Slf4j
 public class KafkaBeanConsumer {
 
-    @KafkaListener(groupId = "group", topics = Topic.BEAN)
+    @KafkaListener(groupId = "beanGroup", topics = Topic.BEAN)
     public void consumer(ConsumerRecord<String, Object> record) {
         System.out.println("消费者收到消息:" + JSON.parseObject(record.value().toString(), Programmer.class));
     }
diff --git a/spring-boot/spring-boot-kafka/src/main/java/com/heibaiying/springboot/consumer/KafkaSimpleConsumer.java b/spring-boot/spring-boot-kafka/src/main/java/com/heibaiying/springboot/consumer/KafkaSimpleConsumer.java
index 55d722c..dfb191b 100644
--- a/spring-boot/spring-boot-kafka/src/main/java/com/heibaiying/springboot/consumer/KafkaSimpleConsumer.java
+++ b/spring-boot/spring-boot-kafka/src/main/java/com/heibaiying/springboot/consumer/KafkaSimpleConsumer.java
@@ -2,6 +2,7 @@ package com.heibaiying.springboot.consumer;
 
 import com.heibaiying.springboot.constant.Topic;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.kafka.support.KafkaHeaders;
@@ -18,8 +19,9 @@ import org.springframework.stereotype.Component;
 public class KafkaSimpleConsumer {
 
     // 简单消费者
-    @KafkaListener(groupId = "group",topics = Topic.SIMPLE)
-    public void consumer1_1(ConsumerRecord<String, Object> record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
+    @KafkaListener(groupId = "simpleGroup", topics = Topic.SIMPLE)
+    public void consumer1_1(ConsumerRecord<String, Object> record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,Consumer consumer) {
         System.out.println("消费者收到消息:" + record.value() + "; topic:" + topic);
+        consumer.commitSync();
     }
 }
diff --git a/spring-boot/spring-boot-kafka/src/main/resources/application.yml b/spring-boot/spring-boot-kafka/src/main/resources/application.yml
index f358f03..cc921dd 100644
--- a/spring-boot/spring-boot-kafka/src/main/resources/application.yml
+++ b/spring-boot/spring-boot-kafka/src/main/resources/application.yml
@@ -1,36 +1,35 @@
 spring:
   kafka:
-    # �Զ��ŷָ��ĵ�ַ�б������ڽ�����Kafka��Ⱥ�ij�ʼ����(kafka Ĭ�ϵĶ˿ں�Ϊ9092)
+    # 以逗号分隔的地址列表，用于建立与Kafka集群的初始连接(kafka 默认的端口号为9092)
     bootstrap-servers: 127.0.0.1:9092
     producer:
-      # �����������Ϣ�ط��Ĵ�����
+      # 发生错误后，消息重发的次数。
       retries: 0
-      #���ж����Ϣ��Ҫ�����͵�ͬһ������ʱ�������߻�����Ƿ���ͬһ��������ò���ָ����һ�����ο���ʹ�õ��ڴ��С�������ֽ������㡣
+      #当有多个消息需要被发送到同一个分区时，生产者会把它们放在同一个批次里。该参数指定了一个批次可以使用的内存大小，按照字节数计算。
       batch-size: 16384
-      # �����������ڴ滺�����Ĵ�С��
+      # 设置生产者内存缓冲区的大小。
       buffer-memory: 33554432
-      # �������л���ʽ
+      # 键的序列化方式
       key-serializer: org.apache.kafka.common.serialization.StringSerializer
-      # ֵ�����л���ʽ
+      # 值的序列化方式
       value-serializer: org.apache.kafka.common.serialization.StringSerializer
-      # acks=0 �� �������ڳɹ�д����Ϣ֮ǰ����ȴ��κ����Է���������Ӧ��
-      # acks=1 �� ֻҪ��Ⱥ������ڵ��յ���Ϣ�������߾ͻ��յ�һ�����Է������ɹ���Ӧ��
-      # acks=all ��ֻ�е����в��븴�ƵĽڵ�ȫ���յ���Ϣʱ�������߲Ż��յ�һ�����Է������ijɹ���Ӧ��
-      acks: all
+      # acks=0 ： 生产者在成功写入消息之前不会等待任何来自服务器的响应。
+      # acks=1 ： 只要集群的首领节点收到消息，生产者就会收到一个来自服务器成功响应。
+      # acks=all ：只有当所有参与复制的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应。
+      acks: 1
     consumer:
-      # �Զ��ύ��ʱ���� ��spring boot 2.X �汾��������õ���ֵ������ΪDuration ��Ҫ�����ض��ĸ�ʽ����1S,1M,2H,5D
+      # 自动提交的时间间隔 在spring boot 2.X 版本中这里采用的是值的类型为Duration 需要符合特定的格式，如1S,1M,2H,5D
       auto-commit-interval: 1S
-      # ������ָ�����������ڶ�ȡһ��û��ƫ�����ķ�������ƫ������Ч������¸����δ�����
-      # latest��Ĭ��ֵ����ƫ������Ч������£������߽������µļ�¼��ʼ��ȡ���ݣ�������������֮�����ɵļ�¼��
-      # earliest ����ƫ������Ч������£������߽�����ʼλ�ö�ȡ�����ļ�¼
+      # 该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下该作何处理：
+      # latest（默认值）在偏移量无效的情况下，消费者将从最新的记录开始读取数据（在消费者启动之后生成的记录）
+      # earliest ：在偏移量无效的情况下，消费者将从起始位置读取分区的记录
       auto-offset-reset: earliest
-      # �Ƿ��Զ��ύƫ������Ĭ��ֵ��true,Ϊ�˱�������ظ����ݺ����ݶ�ʧ�����԰�������Ϊfalse,Ȼ���ֶ��ύƫ����
-      enable-auto-commit: true
-      # ���ķ����л���ʽ
+      # 是否自动提交偏移量，默认值是true,为了避免出现重复数据和数据丢失，可以把它设置为false,然后手动提交偏移量
+      enable-auto-commit: false
+      # 键的反序列化方式
       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
-      # ֵ�ķ����л���ʽ
+      # 值的反序列化方式
       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
     listener:
-      # �����������������е��߳�����
+      # 在侦听器容器中运行的线程数。
       concurrency: 5
-