未调用Kafka侦听器方法。消费者不消费。

Question

<bean id="kafkaConsumerProperties" class="com.azuga.kafka.listeners.KafkaConsumerProperties">
    <constructor-arg type="java.lang.String" value="127.0.0.1:9092" />
    <constructor-arg type="java.lang.String" value="tdm-group" />
    <constructor-arg type="java.lang.String" value="dbStreamer.azuga.tripDriverMapping" />
</bean>
<bean id="kafkaListenerConfig" class="com.azuga.kafka.listeners.KafkaListenerConfig">
    <property name="kafkaConsumerProperties" ref="kafkaConsumerProperties" />
</bean>
<bean id="kafkaContainerFactory" class="com.azuga.kafka.listeners.KafkaListenerContainerFactory"
    factory-method="kafkaContainerFactory">
</bean>

这是创建ListenerContainerFactory的类

@EnableKafka
public class KafkaListenerContainerFactory {

public static ConcurrentKafkaListenerContainerFactory<String, String> kafkaContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConcurrency(1);
    factory.setConsumerFactory(consumerFactory());
    factory.getContainerProperties().setPollTimeout(3000);
    return factory;
}

@SuppressWarnings("unchecked")
public static ConsumerFactory<String, String> consumerFactory() {
    return new DefaultKafkaConsumerFactory<>(KafkaListenerConfig.consumerProps(),
            KafkaListenerConfig.stringKeyDeserializer(), KafkaListenerConfig.stringKeyDeserializer());
}

}

这是我用@KafKalistener注释的Listener类

package com.azuga.kafka.listeners;

import org.springframework.kafka.annotation.KafkaListener;
public class Listener {

@KafkaListener(topics = "dbStreamer.azuga.tripDriverMapping")
public void onMessage(String message) {
    System.out.println(message.toString());
}
}

这是KafkaListenerConfig类，它接受引导服务器、主题名称等。

@EnableKafka
public class KafkaListenerConfig {

private static KafkaConsumerProperties kafkaConsumerProperties;

public void setKafkaConsumerProperties(KafkaConsumerProperties kafkaConsumerProperties) {
    this.kafkaConsumerProperties = kafkaConsumerProperties;
}

public static Map<String, Object> consumerProps() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConsumerProperties.getBootstrap());
    props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaConsumerProperties.getGroup());
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
    props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100");
    props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
    return props;
}

public static Deserializer stringKeyDeserializer() {
    return new StringDeserializer();
}

}

Answer 1

您的应用程序的配置有点不寻常。

但是，我假设您忽略了@enablekafka是关于@configuration类的。因此，根据Spring Framework文档，您必须使用AnnotationConfigWebApplicationContext类：

* {@link org.springframework.web.context.WebApplicationContext WebApplicationContext}
 * implementation which accepts annotated classes as input - in particular
 * {@link org.springframework.context.annotation.Configuration @Configuration}-annotated
 * classes, but also plain {@link org.springframework.stereotype.Component @Component}
 * classes and JSR-330 compliant classes using {@code javax.inject} annotations. Allows
 * for registering classes one by one (specifying class names as config location) as well
 * as for classpath scanning (specifying base packages as config location).

不幸的是，这不能仅用于简单的XML配置。

Spring Kafka没有为XML定义提供任何钩子。