基础理念对于我们有abc3台机器那么我们的消息会被消费3次可能会也可能不会这取决于你的配置和策略。消费者组机制Kafka 使用消费组Consumer Group来确保每个消息只会被每个消费者组中的一个消费者消费一次。分区分配Kafka 主题可以分为多个分区Partitions每个分区只能由一个消费者组中的一个消费者消费。如果新增1台机器那么他的偏移量从0开始在kafka中消费组都会维护自己的偏移量offset以此来记录消费的消息位置而当新增1台机器时会根据分区分配策略比如范围分配、轮询分配这就有2种情况可能加入旧分区也可能加入新分区具体可以配置下策略参数auto.offset.reset对应的值如下earliest: automatically reset the offset to the earliest offsetlatest: automatically reset the offset to the latest offsetnone: throw exception to the consumer if no previous offset is found for the consumer’s groupanything else: throw exception to the consumer.本地调试所以当我们在调试kafka消费逻辑的时候可能由于消费逻辑写的不对改完代码需要重新测重新去造1条消息还是你会怎么去做了解了前面的原理我们改消费组是不可行的他获取的是最新的偏移量无法实现复用之前造的某条数据特别是我们有不同逻辑把每种类型的消息都重新推一波这不仅麻烦而且也容易出错是否可以直接复用之前的消息准确处理我们可以先了解下KafkaListener里面的一些配置参数具体如下KafkaListener(topicPartitions{TopicPartition(topicyourTopic,partitionOffsets{PartitionOffset(partition指定分区,initialOffset初始偏移量)})})publicvoidforlanConsumer(ConsumerRecordString,Stringrecord){StringmessageStrrecord.value();log.info(测试触达记录消费:offset {}, res {},record.offset(),messageStr);}上面只是指定了我们从什么偏移量开始消费如果要限制范围可以在代码里面加限制KafkaListener(topicPartitions{TopicPartition(topicyourTopic,partitionOffsets{PartitionOffset(partition指定分区,initialOffset初始偏移量)})})publicvoidforlanConsumer(ConsumerRecordString,Stringrecord){if(record.offset()结束偏移量)return;StringmessageStrrecord.value();log.info(测试触达记录消费:offset {}, res {},record.offset(),messageStr);}测试或正式环境调试上面只适合本地场景如果是线上环境我们本地一般是没有权限连接监听的那么可以怎么做其实也能做只不过需要通过接口去拉取处理AutowiredprivateKafkaConfigkafkaConfig;publicvoidreconsumeMessage(Stringtopic,intpartition,longoffset){ConsumerFactoryInteger,StringconsumerFactorykafkaConfig.consumerFactory();MapString,ObjectconfigurationPropertiesconsumerFactory.getConfigurationProperties();MapString,ObjectcustomPropsnewHashMap();customProps.put(ConsumerConfig.GROUP_ID_CONFIG,reconsume-temp-group-UUID.randomUUID());customProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,earliest);// 防止 offset 不存在时报错customProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);customProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,1);// 控制每次 poll 只拉取一条// 复用 kafkaConfig 中的基础配置MapString,ObjectpropsnewHashMap(configurationProperties);props.putAll(customProps);try(ConsumerString,StringconsumernewKafkaConsumer(props)){TopicPartitiontopicPartitionnewTopicPartition(topic,partition);// 分配分区并定位偏移量consumer.assign(Collections.singletonList(topicPartition));consumer.seek(topicPartition,offset);// 拉取消息设置超时时间ConsumerRecordsString,Stringrecordsconsumer.poll(Duration.ofSeconds(5));for(ConsumerRecordString,Stringrecord:records){if(record.offset()offset){if(Objects.equals(topic,KafkaTopic.Forlan_MESSAGE_NOTIFY)){// 调用KafkaListener的方法执行逻辑forlanConsumer.userConsumer(record);}break;}}}}项目中如果没有配置kafkaConfig也可以自定义一个只要能拿到连接就行privateMapString,ObjectconsumerConfigs(){MapString,ObjectpropsnewHashMap();props.put(bootstrap.servers,this.bootstrapServers);props.put(group.id,this.groupid);props.put(enable.auto.commit,this.autoCommit);props.put(auto.commit.interval.ms,this.interval);props.put(session.timeout.ms,this.timeout);props.put(key.deserializer,this.keyDeserializer);props.put(value.deserializer,this.valueDeserializer);props.put(auto.offset.reset,this.offsetReset);props.put(max.poll.records,this.maxPollRecords);returnprops;}