Spring 4.1.0.RELEASE能否集成Kafka？遗留项目需短期方案

方案1：降级spring-kafka到兼容版本 #

spring-kafka 1.3.x系列依赖Spring 4.2+，而你的项目基于Spring 4.1.0，因此需要降级到spring-kafka 1.2.10.RELEASE——该版本官方兼容Spring 4.1.x（最低要求Spring 4.1.9.RELEASE，若你的4.1.0存在小版本兼容问题，可考虑升级到4.1.x最新小版本，这属于同一主版本内的微调，通常不会影响原有业务）。

依赖配置（Maven） #

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>1.2.10.RELEASE</version>
</dependency>
<!-- 匹配spring-kafka 1.2.x的kafka-clients版本 -->
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>0.10.2.2</version>
</dependency>

核心配置示例（XML） #

生产者配置

<bean id="kafkaProducerFactory" class="org.springframework.kafka.core.DefaultKafkaProducerFactory">
    <constructor-arg>
        <map>
            <entry key="bootstrap.servers" value="你的Kafka集群地址"/>
            <entry key="key.serializer" value="org.apache.kafka.common.serialization.StringSerializer"/>
            <entry key="value.serializer" value="org.apache.kafka.common.serialization.StringSerializer"/>
        </map>
    </constructor-arg>
</bean>

<bean id="kafkaTemplate" class="org.springframework.kafka.core.KafkaTemplate">
    <constructor-arg ref="kafkaProducerFactory"/>
    <property name="defaultTopic" value="你的目标Topic"/>
</bean>

消费者配置

<bean id="kafkaConsumerFactory" class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
    <constructor-arg>
        <map>
            <entry key="bootstrap.servers" value="你的Kafka集群地址"/>
            <entry key="group.id" value="你的消费组ID"/>
            <entry key="key.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer"/>
            <entry key="value.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer"/>
            <entry key="auto.offset.reset" value="earliest"/>
        </map>
    </constructor-arg>
</bean>

<!-- 自定义消息监听器 -->
<bean id="customMessageListener" class="com.yourpackage.CustomMessageListener"/>

<bean id="containerProperties" class="org.springframework.kafka.listener.ContainerProperties">
    <constructor-arg value="你的目标Topic"/>
    <property name="messageListener" ref="customMessageListener"/>
</bean>

<bean id="kafkaListenerContainer" class="org.springframework.kafka.listener.KafkaMessageListenerContainer" 
      init-method="start" destroy-method="stop">
    <constructor-arg ref="kafkaConsumerFactory"/>
    <constructor-arg ref="containerProperties"/>
</bean>

消息监听器实现

public class CustomMessageListener implements MessageListener<String, String> {
    @Override
    public void onMessage(ConsumerRecord<String, String> record) {
        // 业务处理逻辑
        System.out.printf("收到消息：Topic=%s, Offset=%d, Value=%s%n", 
                          record.topic(), record.offset(), record.value());
    }
}

方案2：手动封装原生Kafka客户端 #

如果不想依赖spring-kafka，可以直接使用Kafka原生客户端，通过Spring的生命周期接口管理客户端的初始化与销毁，完全脱离spring-kafka的Spring版本限制。

依赖配置（Maven） #

选择与Spring 4.1.x兼容性较好的kafka-clients版本，比如0.9.0.1：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>0.9.0.1</version>
</dependency>

生产者Bean实现 #

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import java.util.Properties;

public class KafkaProducerBean implements InitializingBean, DisposableBean {
    private Producer<String, String> producer;
    private String bootstrapServers;

    public void setBootstrapServers(String bootstrapServers) {
        this.bootstrapServers = bootstrapServers;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        Properties props = new Properties();
        props.put("bootstrap.servers", bootstrapServers);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        producer = new KafkaProducer<>(props);
    }

    // 发送消息方法
    public void send(String topic, String key, String value) {
        producer.send(new ProducerRecord<>(topic, key, value));
    }

    @Override
    public void destroy() throws Exception {
        if (producer != null) {
            producer.close();
        }
    }
}

消费者Bean实现 #

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class KafkaConsumerBean implements InitializingBean, DisposableBean {
    private Consumer<String, String> consumer;
    private ExecutorService executorService;
    private String bootstrapServers;
    private String groupId;
    private String topic;

    public void setBootstrapServers(String bootstrapServers) {
        this.bootstrapServers = bootstrapServers;
    }

    public void setGroupId(String groupId) {
        this.groupId = groupId;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        Properties props = new Properties();
        props.put("bootstrap.servers", bootstrapServers);
        props.put("group.id", groupId);
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("auto.offset.reset", "earliest");
        
        consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList(topic));

        // 启动线程轮询消息
        executorService = Executors.newSingleThreadExecutor();
        executorService.submit(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                ConsumerRecords<String, String> records = consumer.poll(100);
                records.forEach(record -> {
                    // 业务处理逻辑
                    System.out.printf("收到消息：Topic=%s, Offset=%d, Value=%s%n", 
                                      record.topic(), record.offset(), record.value());
                });
            }
        });
    }

    @Override
    public void destroy() throws Exception {
        if (consumer != null) {
            consumer.close();
        }
        if (executorService != null) {
            executorService.shutdownNow();
        }
    }
}

Spring XML配置 #

<bean id="kafkaProducer" class="com.yourpackage.KafkaProducerBean">
    <property name="bootstrapServers" value="你的Kafka集群地址"/>
</bean>

<bean id="kafkaConsumer" class="com.yourpackage.KafkaConsumerBean">
    <property name="bootstrapServers" value="你的Kafka集群地址"/>
    <property name="groupId" value="你的消费组ID"/>
    <property name="topic" value="你的目标Topic"/>
</bean>

方案3：补全缺失的Spring类（应急不推荐） #

错误ClassNotFoundException: org.springframework.core.MethodIntrospector是因为该类从Spring 4.2才开始提供，你可以临时从Spring 4.2.x的spring-core模块中提取该类及依赖的相关类（如MethodIntrospector.MetadataLookup、AnnotationFilter）的源码，复制到项目的org.springframework.core包下。

注意：这种方式会与现有Spring版本存在潜在冲突，可能引发未知问题，仅作为短期应急方案，长期来看仍需规划Spring版本升级或改用前两种方案。

内容的提问来源于stack exchange，提问作者P.Büth