很早之前就使用了springboot + kafka组合配置，但是之前使用的spring-kafka（1.1.7）版本较低，所以只能通过 spring.kafka.consumer.group-id=default_consumer_group 或者 propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, “default_consumer_group”);的形式配置一个默认消组，当然理论上这也是没有问题的，但是如果你定义的topic数量过多且并发消费比较大，只有一个消费组的配置方式就会暴露出很多问题，其中主要的一个问题便是每个topic分区的offset偏移量问题（在大并发下会出现offset异常问题），因为他们都保存在同一个消费组中。

直到后来发布了spring-kafka 1.3.x的版本后，增加了groupId的属性，非常方便的帮助我们解决了实现每个topic自定义一个消费组的问题，我们再也不用共用一个消费组了。

接下来通过代码演示看是否如我们的期望一样：

pom依赖

1. <parent>
2. <groupId>org.springframework.boot</groupId>
3. <artifactId>spring-boot-starter-parent</artifactId>
4. <version>1.5.10.RELEASE</version>
5. <relativePath/> <!– lookup parent from repository –>
6. </parent>
8. <properties>
9. <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
10. <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
11. <java.version>1.8</java.version>
12. </properties>
14. <dependencies>
15. <dependency>
16. <groupId>org.springframework.boot</groupId>
17. <artifactId>spring-boot-starter-web</artifactId>
18. </dependency>
19. <!– https://mvnrepository.com/artifact/org.springframework.kafka/spring-kafka –>
20. <dependency>
21. <groupId>org.springframework.kafka</groupId>
22. <artifactId>spring-kafka</artifactId>
23. <version>1.3.5.RELEASE</version>
24. </dependency>
26. <dependency>
27. <groupId>org.springframework.boot</groupId>
28. <artifactId>spring-boot-starter-test</artifactId>
29. <scope>test</scope>
30. </dependency>
32. <!–引入elasticsearch–>
33. <dependency>
34. <groupId>org.springframework.boot</groupId>
35. <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
36. </dependency>
37. </dependencies>
39. <build>
40. <plugins>
41. <plugin>
42. <groupId>org.springframework.boot</groupId>
43. <artifactId>spring-boot-maven-plugin</artifactId>
44. </plugin>
45. </plugins>
46. </build>

application.properties

1. server.port=10087
2. spring.application.name=example
3. #topic
4. spring.kafka.bootstrap-servers=10.0.2.22:9092
5. kafka.test.topic=TEST_TOPIC
8. #es
9. spring.data.elasticsearch.cluster-name=elasticsearch
10. spring.data.elasticsearch.cluster-nodes=10.0.2.23:9300
11. #spring.data.elasticsearch.cluster-nodes=10.0.2.22:9300

生产者：

1. /**
2. * @author xiaofeng
3. * @version V1.0
4. * @title: TestKafkaSender.java
5. * @package: com.example.demo.kafka.sender
6. * @description: kafka生产者
7. * @date 2018/4/2 0002 下午 3:31
8. */
9. @Component
10. public class TestKafkaSender {
11. @Autowired
12. private KafkaTemplate kafkaTemplate;
14. @Value(“${kafka.test.topic}”)
15. String testTopic;
17. public void sendTest(String msg){
18. kafkaTemplate.send(testTopic, msg);
19. }
20. }

消费者1：

1. /**
2. * @author xiaofeng
3. * @version V1.0
4. * @title: TestKafkaConsumer2.java
5. * @package: com.example.demo.kafka.consumer
6. * @description: kafka消费者
7. * @date 2018/4/2 0002 下午 3:31
8. */
9. @Component
10. public class TestKafkaConsumer {
12. Logger logger = LoggerFactory.getLogger(getClass());
14. /**
15. * topics: 配置消费topic，以数组的形式可以配置多个
16. * groupId: 配置消费组为”xiaofeng1“
17. *
18. * @param message
19. */
20. @KafkaListener(topics = {“${kafka.test.topic}”},groupId = “xiaofeng1”)
21. public void consumer(String message) {
22. logger.info(“groupId = xiaofeng1, message = “ + message);
23. }
25. }

消费者2：

1. /**
2. * @author xiaofeng
3. * @version V1.0
4. * @title: TestKafkaConsumer2.java
5. * @package: com.example.demo.kafka.consumer
6. * @description: kafka消费者
7. * @date 2018/4/2 0002 下午 3:31
8. */
9. @Component
10. public class TestKafkaConsumer2 {
12. Logger logger = LoggerFactory.getLogger(getClass());
14. /**
15. * topics: 配置消费topic，以数组的形式可以配置多个
16. * groupId: 配置消费组为”xiaofeng2“
17. *
18. * @param message
19. */
20. @KafkaListener(topics = {“${kafka.test.topic}”}, groupId = “xiaofeng2”)
21. public void consumer(String message) {
22. logger.info(“groupId = xiaofeng2, message = “ + message);
23. }
25. }

测试类：

1. @Autowired
2. TestKafkaSender sender;
4. @Test
5. public void send() {
6. for (int i = 0; i < Integer.MAX_VALUE; i++) {
7. logger.info(“send message = “ + i);
8. sender.sendTest(i + “”);
9. try {
10. Thread.sleep(1000);
11. } catch (InterruptedException e) {
12. e.printStackTrace();
13. }
14. }
15. }

运行效果：