springboot kafka group.id多消费组配置

曾经终败给现在 2022-05-12 18:34 2050阅读 0赞

很早之前就使用了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/> 
    	</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>
    		
    		<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>
    
    		
    		<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();
                }
            }
        }

运行效果：

![70][]

[70]: /images/20220504/a4be990396c0484a824d27a41e31c5bd.png