我是第一次使用 Spring Kafka,我无法在我的消费者代码中使用 Acknowledgement.acknowledge() 方法进行手动提交,如此处所述 https://docs.spring.io/spring-kafka/reference/html/ _reference.html#committing-offsets。我的是 spring-boot 应用程序。如果我不使用手动提交过程,那么我的代码工作正常。但是当我使用 Acknowledgement.acknowledge() 用于手动提交,它显示与 bean 相关的错误。另外,如果我没有正确使用手动提交,请建议我正确的方法。
错误信息:
***************************
APPLICATION FAILED TO START
***************************
Description:
Field ack in Receiver required a bean of type 'org.springframework.kafka.support.Acknowledgment' that could not be found.
Action:
Consider defining a bean of type 'org.springframework.kafka.support.Acknowledgment' in your configuration.
我用谷歌搜索了这个错误,我发现我需要添加@Component,但这已经存在于我的消费者代码中。
我的消费者代码如下所示:Receiver.java
import java.util.concurrent.CountDownLatch;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
@Component
public class Receiver {
@Autowired
public Acknowledgment ack;
private CountDownLatch latch = new CountDownLatch(1);
@KafkaListener(topics = "${kafka.topic.TestTopic}")
public void receive(ConsumerRecord<?, ?> consumerRecord){
System.out.println(consumerRecord.value());
latch.countDown();
ack.acknowledge();
}
}
我的生产者代码如下所示:Sender.java
import java.util.Map;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
@Component
public class Sender {
@Autowired
private KafkaTemplate<String, Map<String, Object>> kafkaTemplate;
public void send(Map<String, Object> map){
kafkaTemplate.send("TestTopic", map);
}
}
编辑1:
我的新消费者代码如下所示:Receiver.java
import java.util.concurrent.CountDownLatch;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
@Component
public class Receiver {
private CountDownLatch latch = new CountDownLatch(1);
@KafkaListener(topics = "${kafka.topic.TestTopic}", containerFactory = "kafkaManualAckListenerContainerFactory")
public void receive(ConsumerRecord<?, ?> consumerRecord, Acknowledgment ack){
System.out.println(consumerRecord.value());
latch.countDown();
ack.acknowledge();
}
}
我也更改了我的配置类:
import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
@Configuration
@EnableKafka
public class ReceiverConfig {
@Value("${kafka.bootstrap-servers}")
private String bootstrapServers;
@Value("${spring.kafka.consumer.group-id}")
private String consumerGroupId;
@Bean
public Map<String, Object> consumerConfigs() throws SendGridException {
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroupId);
return props;
}
@Bean
public ConsumerFactory<String, String> consumerFactory(){
return new DefaultKafkaConsumerFactory<>(consumerConfigs());
}
/*@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(){
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
return factory;
}*/
@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaManualAckListenerContainerFactory(){
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
return factory;
}
@Bean
public Receiver receiver() {
return new Receiver();
}
}
将containerFactory =“kafkaManualAckListenerContainerFactory”添加到我的receive()方法后,我收到以下错误。
***************************
APPLICATION FAILED TO START
***************************
Description:
Parameter 1 of method kafkaListenerContainerFactory in org.springframework.boot.autoconfigure.kafka.KafkaAnnotationDrivenConfiguration required a bean of type 'org.springframework.kafka.core.ConsumerFactory' that could not be found.
- Bean method 'kafkaConsumerFactory' in 'KafkaAutoConfiguration' not loaded because @ConditionalOnMissingBean (types: org.springframework.kafka.core.ConsumerFactory; SearchStrategy: all) found bean 'consumerFactory'
Action:
Consider revisiting the conditions above or defining a bean of type 'org.springframework.kafka.core.ConsumerFactory' in your configuration.
对于那些仍在寻找有关手动确认的这些错误的解决方案的人,您不需要指定containerFactory =“kafkaManualAckListenerContainerFactory”,而是只需添加:
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
在返回工厂对象之前添加到接收器配置。
那么你还需要:
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
在消费者配置道具中。
所以最后你的监听器方法可以简单地看起来像:
@KafkaListener(topics = "${spring.kafka.topic}")
private void listen(@Payload String payload, Acknowledgment acknowledgment) {
//Whatever code you want to do with the payload
acknowledgement.acknowledge(); //or even pass the acknowledgment to a different method and acknowledge even later
}
您确实应该遵循文档:
使用手动
时,也可以向听者提供AckMode
;这个示例还展示了如何使用不同的容器工厂。Acknowledgment
@KafkaListener(id = "baz", topics = "myTopic",
containerFactory = "kafkaManualAckListenerContainerFactory")
public void listen(String data, Acknowledgment ack) {
...
ack.acknowledge();
}
确实没有任何地方指出
Acknowledgment
是豆子。因此,适当更改您的 receive()
@KafkaListener
方法签名,并删除可疑 @Autowired
bean 的 Acknowledgment
- 它只是不存在,因为该对象是每个收到消息的一部分(标头)。
那些使用 Spring Boot 应用程序的人,只需将以下内容添加到您的 application.yml (或特定于环境的文件)中。
spring:
kafka:
listener:
ack-mode: manual
上述更改将使
Acknowledgment ack
参数在内部可用receive(ConsumerRecord<?, ?> consumerRecord, Acknowledgment ack)
自动方法。
Java代码:
@Service
public class ServiceCallKafkaListener {
@KafkaListener(id = "listenerId",
groupId = "groupListenerId",
topics = "topicName")
public void listenServiceCall(@Payload String message,
Acknowledgment acknowledgment) {
//here is your logic for message processing
boolean logicForMessageProcessingCompleted = true;
if (logicForMessageProcessingCompleted) {
//manual commit
acknowledgment.acknowledge();
}
}
}
应用程序.属性
#type of acknowledgment mode
spring.kafka.listener.ack-mode=MANUAL_IMMEDIATE
#property that turns off auto commit
spring.kafka.consumer.enable-auto-commit=false