【发布时间】：2019-10-14 17:18:05
【问题描述】：
我正在尝试在从主题中阅读后提交一条消息。我已经按照这个链接 (https://www.confluent.io/blog/apache-kafka-spring-boot-application) 创建了一个带有 spring 的 Kafka 消费者。通常它工作得很好，消费者收到消息并等到另一个人进入队列。但问题是，当我处理这些消息时，它需要很长时间（大约 10 分钟），kafka 队列认为消息没有被消费（提交）并且消费者一次又一次地读取它。我不得不说，当我的处理时间少于 5 分钟时，它运行良好，但当它持续更长时间时，它不会提交消息。
我一直在寻找一些答案，但这对我没有帮助，因为我没有使用相同的源代码（当然还有不同的结构）。我尝试发送异步方法并异步提交消息，但我失败了。 一些来源是：
Spring Boot Kafka: Commit cannot be completed since the group has already rebalanced
https://dzone.com/articles/kafka-clients-at-most-once-at-least-once-exactly-o
Kafka 0.10 Java consumer not reading message from topic
https://github.com/confluentinc/confluent-kafka-dotnet/issues/470
主要课程在这里：
@SpringBootApplication
@EnableAsync
public class SpringBootKafkaApp {
public static void main(String[] args) {
SpringApplication.run(SpringBootKafkaApp .class, args);
}
消费者类（我需要提交消息的地方）
@Service
public class Consumer {
@Autowired
AppPropert prop;
Consumer cons;
@KafkaListener(topics = "${app.topic.pro}", groupId = "group_id")
public void consume(String message) throws IOException {
/*HERE I MUST CONSUME THE MESSAGE AND COMMIT IT */
Properties props=prope.startProp();//just getting my properties from my config-file
ControllerPRO pro = new ControllerPRO();
List<Future<String>> async= new ArrayList<Future<String>>();//call this method asynchronous, doesn't help me
try {
CompletableFuture<String> ret=pro.processLaunch(message,props);//here I call the process method
/*This works fine when the processLaunch method takes less than 5 minutes,
if it takes longer the consumer will get the same message from the topic and start again with this operation
*/
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
System.out.println("End of consumer method ");
}
}
如何在从队列中读取消息后立即提交消息。
我想确保在收到消息时立即提交消息。现在，当我在 (System.out.println) 之后完成执行该方法时，将提交该消息。那么谁能告诉我怎么做呢？
-----更新-------
抱歉，回复晚了，但正如@GirishB 所建议的那样，我一直在寻找 GirishB 的配置，但我看不到在哪里可以定义我想从我的配置文件 (applications.yml) 中读取/收听的主题.我看到的所有示例都使用与此类似的结构 (http://tutorials.jenkov.com/java-util-concurrent/blockingqueue.html)。是否有任何选项可以读取在其他服务器中声明的主题？使用类似于 @KafkaListener(topics = "${app.topic.pro}", groupId = "group_id")
=========== 解决方案1 =================================== ======
我遵循了@victor gallet 的建议，并在 oder 中包含了 confumer 属性的声明，以适应 consumer 方法中的“Acknowledgment”对象。我还通过此链接 (https://www.programcreek.com/java-api-examples/?code=SpringOnePlatform2016/grussell-spring-kafka/grussell-spring-kafka-master/s1p-kafka/src/main/java/org/s1p/CommonConfiguration.java) 获取了我用来声明和设置所有属性（consumerProperties、consumerFactory、kafkaListenerContainerFactory）的所有方法。我发现的唯一问题是 “new SeekToCurrentErrorHandler()”声明，因为我遇到了一个错误，目前我无法解决它（如果有人向我解释它会很棒）。
@Service
public class Consumer {
@Autowired
AppPropert prop;
Consumer cons;
@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory();
factory.setConsumerFactory(consumerFactory());
factory.getContainerProperties().setAckOnError(false);
factory.getContainerProperties().setAckMode(AckMode.MANUAL_IMMEDIATE);
//factory.setErrorHandler(new SeekToCurrentErrorHandler());//getting error here despite I've loaded the library
return factory;
}
@Bean
public ConsumerFactory<String, String> consumerFactory() {
return new DefaultKafkaConsumerFactory<>(consumerProperties());
}
@Bean
public Map<String, Object> consumerProperties() {
Map<String, Object> props = new HashMap<>();
Properties propsManu=prop.startProperties();// here I'm getting my porperties file where I retrive the configuration from a remote server (you have to trust that this method works)
//props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.configProperties.getBrokerAddress());
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, propsManu.getProperty("bootstrap-servers"));
//props.put(ConsumerConfig.GROUP_ID_CONFIG, "s1pGroup");
props.put(ConsumerConfig.GROUP_ID_CONFIG, propsManu.getProperty("group-id"));
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 15000);
//props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, propsManu.getProperty("key-deserializer"));
//props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, propsManu.getProperty("value-deserializer"));
return props;
}
@KafkaListener(topics = "${app.topic.pro}", groupId = "group_id")
public void consume(String message) throws IOException {
/*HERE I MUST CONSUME THE MESSAGE AND COMMIT IT */
acknowledgment.acknowledge();// commit immediately
Properties props=prop.startProp();//just getting my properties from my config-file
ControllerPRO pro = new ControllerPRO();
List<Future<String>> async= new ArrayList<Future<String>>();//call this method asynchronous, doesn't help me
try {
CompletableFuture<String> ret=pro.processLaunch(message,props);//here I call the process method
/*This works fine when the processLaunch method takes less than 5 minutes,
if it takes longer the consumer will get the same message from the topic and start again with this operation
*/
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
System.out.println("End of consumer method ");
}
}
``````````````````````````````````````````````````````````
【问题讨论】：
-
如果您的处理时间很长并且只想在处理成功后提交，请考虑调整这些设置：
max.poll.interval.ms、max.poll.interval.ms和max.poll.records参见kafka.apache.org/documentation
标签： java spring apache-kafka