KafKa手动提交问题描述

KafKa手动提交offset问题描述

在分布式架构，我在XXJOB来进行调用接口产生数据，将数据提交KafKa进行存储；然后将KafKa中数据拿出来消费，在处理KafKa出来的数据时，调用另外一个服务时，正好另外一个服务挂了，导致手动提交offset失败，造成数据的重复消费；

@KafkaListener(topics = "xxx",groupId = "consumer01")
    public void listen(ConsumerRecord record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic, Acknowledgment ack) {
   
   		/**
   			代码块（其中有调用别的服务接口）
   		**/
   		// 手动提交offset
        ack.acknowledge();
    }

问题解决

解决方案就是将可能有问题的代码使用try…catch进行捕获异常，然后在finally进行手动提交offset

@KafkaListener(topics = "xxx",groupId = "consumer01")
    public void listen(ConsumerRecord record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic, Acknowledgment ack) {
   
    	try{
   
	    	/**
	   			代码块（其中有调用别的服务接口）
	   		**/
    	}catch(Exception e){
   
    		
    	}finally{
   
    		// 手动提交offset
        	ack.acknowledge();
    	}
   		
   		
    }

解决方案仅供参考，不喜勿喷。