Spring Kafka Without spring 引导消费者不消费消息

Question

Consumer 使用 Spring 的 JavaConfig class 如下：

@Configuration
@EnableKafka
public class KafkaConfig {

    public static final String TOPIC = "test-1";
    private String bootstrapServers = "localhost:9092";

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return props;
    }

使用@KafkaListener注解的Kafka主题监听器如下：

@Component
public class MessageListener {

    private static final Logger LOGGER = LoggerFactory.getLogger(MessageListener.class);

    @KafkaListener(topics = KafkaConfig.TOPIC)
    public void handle(ConsumerRecord<?, ?> cr) {
        LOGGER.info("Message: "+cr.key()+"="+cr.value());
    }
}

我的 pom 包含依赖项：

 <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>2.2.6.RELEASE</version>
        </dependency>

现在，当我打包到 war 并部署到 tomcat 时，它不会显示任何错误，甚至在调试模式下也不会只是部署 war 而什么也没有。

如果我遗漏了一些触发 kafkalistner 的配置，请帮助我理解。

谢谢加里，我添加了 context.xml 和 web.xml 但我在下面的评论中提到了 bean 错误

Web.XML

<web-app id="WebApp_ID" version="2.4"
    xmlns="http://java.sun.com/xml/ns/j2ee" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://java.sun.com/xml/ns/j2ee 
    http://java.sun.com/xml/ns/j2ee/web-app_2_4.xsd">

    <display-name>Spring Web MVC Application</display-name>


    <context-param>
        <param-name>contextConfigLocation</param-name>
        <param-value>WEB-INF/context.xml</param-value>
    </context-param>

    <listener>
        <listener-class>org.springframework.web.context.ContextLoaderListener</listener-class>
    </listener>

</web-app>

Context.XML

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:context="http://www.springframework.org/schema/context" xmlns:p="http://www.springframework.org/schema/p"
    xmlns:util="http://www.springframework.org/schema/util"
    xsi:schemaLocation="http://www.springframework.org/schema/beans
     http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
     http://www.springframework.org/schema/context
     http://www.springframework.org/schema/context/spring-context-3.0.xsd
     http://www.springframework.org/schema/util
     http://www.springframework.org/schema/util/spring-util-3.0.xsd">

    <context:component-scan base-package="com.mkyong.common.controller.*" />
    <context:annotation-config></context:annotation-config>

</beans>

Answer 1

在 war 中部署时，您需要通过 web.xml bootstrap 应用程序上下文。参见 https://docs.spring.io/spring/docs/5.2.5.RELEASE/spring-framework-reference/web.html#web-integration-common

<listener>
    <listener-class>org.springframework.web.context.ContextLoaderListener</listener-class>
</listener>

...
<context-param>
    <param-name>contextConfigLocation</param-name>
    <param-value>/WEB-INF/context.xml</param-value>
</context-param>

在 context.xml 添加 <context:component-scan/>.

见https://docs.spring.io/spring/docs/5.2.5.RELEASE/spring-framework-reference/core.html#beans-scanning-autodetection

Answer 2

问题是 spring 默认情况下使用 XmlWebApplicationContext，所以我必须明确告诉 Spring 通过使用 java 类作为 bean 定义的输入来为 Web 应用程序创建应用程序上下文xml 个文件。

下面是对 context.xml 所做的更改。这对我有用。

`

<listener>
        <listener-class>org.springframework.web.context.ContextLoaderListener</listener-class>
    </listener>

<context-param>
        <param-name>contextClass</param-name>
        <param-value>org.springframework.web.context.support.AnnotationConfigWebApplicationContext</param-value>
    </context-param>
    <context-param>
    <param-name>contextConfigLocation</param-name>
    <param-value>com.test.common.controller.KafkaConfig</param-value>
</context-param>`

Spring Kafka Without spring 引导消费者不消费消息

Spring Kafka Without spring boot consumer not consuming messages

spring

spring-kafka