很早之前就使用了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依赖
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>1.5.10.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- https://mvnrepository.com/artifact/org.springframework.kafka/spring-kafka -->
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>1.3.5.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<!--引入elasticsearch-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-elasticsearch</artifactId>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
application.properties
server.port=10087
spring.application.name=example
#topic
spring.kafka.bootstrap-servers=10.0.2.22:9092
kafka.test.topic=TEST_TOPIC
#es
spring.data.elasticsearch.cluster-name=elasticsearch
spring.data.elasticsearch.cluster-nodes=10.0.2.23:9300
#spring.data.elasticsearch.cluster-nodes=10.0.2.22:9300生产者：
/**
* @author xiaofeng
* @version V1.0
* @title: TestKafkaSender.java
* @package: com.example.demo.kafka.sender
* @description: kafka生产者
* @date 2018/4/2 0002 下午 3:31
*/
@Component
public class TestKafkaSender {
@Autowired
private KafkaTemplate kafkaTemplate;
@Value("${kafka.test.topic}")
String testTopic;
public void sendTest(String msg){
kafkaTemplate.send(testTopic, msg);
}
}消费者1：
/**
* @author xiaofeng
* @version V1.0
* @title: TestKafkaConsumer2.java
* @package: com.example.demo.kafka.consumer
* @description: kafka消费者
* @date 2018/4/2 0002 下午 3:31
*/
@Component
public class TestKafkaConsumer {
Logger logger = LoggerFactory.getLogger(getClass());
/**
* topics: 配置消费topic，以数组的形式可以配置多个
* groupId: 配置消费组为”xiaofeng1“
*
* @param message
*/
@KafkaListener(topics = {"${kafka.test.topic}"},groupId = "xiaofeng1")
public void consumer(String message) {
logger.info("groupId = xiaofeng1, message = " + message);
}
}消费者2：
/**
* @author xiaofeng
* @version V1.0
* @title: TestKafkaConsumer2.java
* @package: com.example.demo.kafka.consumer
* @description: kafka消费者
* @date 2018/4/2 0002 下午 3:31
*/
@Component
public class TestKafkaConsumer2 {
Logger logger = LoggerFactory.getLogger(getClass());
/**
* topics: 配置消费topic，以数组的形式可以配置多个
* groupId: 配置消费组为”xiaofeng2“
*
* @param message
*/
@KafkaListener(topics = {"${kafka.test.topic}"}, groupId = "xiaofeng2")
public void consumer(String message) {
logger.info("groupId = xiaofeng2, message = " + message);
}
}
测试类：
@Autowired
TestKafkaSender sender;
@Test
public void send() {
for (int i = 0; i < Integer.MAX_VALUE; i++) {
logger.info("send message = " + i);
sender.sendTest(i + "");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}运行效果：
