传统Spring XML配置应用集成Confluent Kafka Consumer方案咨询

1. 依赖配置 #

先确保项目引入Spring Kafka及Confluent Kafka客户端依赖（以Maven为例），版本需与你的Spring版本、Confluent Kafka集群版本兼容：

<dependencies>
    <!-- Spring Kafka 核心依赖 -->
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
        <version>2.8.10</version> <!-- 适配Spring 5.x版本 -->
    </dependency>
    <!-- Confluent Kafka 客户端（需Confluent扩展特性时添加） -->
    <dependency>
        <groupId>io.confluent</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>7.3.3</version> <!-- 与Confluent集群版本匹配 -->
    </dependency>
    <!-- Spring Context 核心依赖（未引入则添加） -->
    <dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-context</artifactId>
        <version>5.3.24</version>
    </dependency>
</dependencies>

2. Spring XML核心配置 #

创建spring-kafka-config.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:kafka="http://www.springframework.org/schema/kafka"
       xsi:schemaLocation="
        http://www.springframework.org/schema/beans
        http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/kafka
        http://www.springframework.org/schema/kafka/spring-kafka.xsd">

    <!-- Kafka 消费者配置参数 -->
    <bean id="consumerProperties" class="java.util.HashMap">
        <constructor-arg>
            <map>
                <!-- Confluent Kafka 集群地址 -->
                <entry key="bootstrap.servers" value="localhost:9092"/>
                <!-- 消费者组ID -->
                <entry key="group.id" value="confluent-consumer-group"/>
                <!-- 自动提交偏移量（按需改为手动提交） -->
                <entry key="enable.auto.commit" value="true"/>
                <entry key="auto.commit.interval.ms" value="1000"/>
                <!-- String类型反序列化器（Avro格式需替换为对应反序列化器） -->
                <entry key="key.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer"/>
                <entry key="value.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer"/>
                <!-- 初始偏移量策略：latest/earliest -->
                <entry key="auto.offset.reset" value="latest"/>
            </map>
        </constructor-arg>
    </bean>

    <!-- Kafka 消费者工厂 -->
    <bean id="consumerFactory" class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
        <constructor-arg ref="consumerProperties"/>
    </bean>

    <!-- 自定义消息监听器：处理接收到的消息 -->
    <bean id="kafkaMessageListener" class="com.yourpackage.KafkaMessageListener"/>

    <!-- 并发消息监听器容器：启动消费者并监听指定Topic -->
    <kafka:listener-container id="kafkaListenerContainer"
                              consumer-factory="consumerFactory"
                              concurrency="3"> <!-- 并发消费线程数，建议不超过Topic分区数 -->
        <kafka:listener id="confluentTopicListener"
                        topics="your-target-topic" <!-- 替换为实际监听的Topic名称 -->
                        ref="kafkaMessageListener"/>
    </kafka:listener-container>

</beans>

3. 实现消息监听器类 #

编写自定义消息处理类，实现Spring Kafka的MessageListener接口，处理业务逻辑：

package com.yourpackage;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.listener.MessageListener;

public class KafkaMessageListener implements MessageListener<String, String> {

    @Override
    public void onMessage(ConsumerRecord<String, String> record) {
        // 解析消息元数据与内容
        String topic = record.topic();
        int partition = record.partition();
        long offset = record.offset();
        String key = record.key();
        String value = record.value();

        // 打印消息信息（调试用）
        System.out.printf("Received message: Topic=%s, Partition=%d, Offset=%d, Key=%s, Value=%s%n",
                topic, partition, offset, key, value);

        // 此处添加你的业务逻辑：比如解析消息、存储数据、调用服务等
    }
}

4. 启动消费者应用 #

编写启动类加载Spring上下文，启动消费者：

package com.yourpackage;

import org.springframework.context.support.ClassPathXmlApplicationContext;

public class KafkaConsumerApp {

    public static void main(String[] args) {
        // 加载Spring XML配置
        ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("spring-kafka-config.xml");
        context.registerShutdownHook(); // 确保应用关闭时优雅停止消费者

        System.out.println("Kafka Consumer started, listening for messages...");

        // 阻塞主线程，保持应用运行
        try {
            Thread.currentThread().join();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

关键注意事项 #

• 版本兼容：务必保证spring-kafka、kafka-clients及Spring核心版本相互匹配，避免版本冲突导致的异常。
• Avro格式支持：若Confluent Producer发送的是Avro格式消息，需替换反序列化器为io.confluent.kafka.serializers.KafkaAvroDeserializer，并添加schema.registry.url配置项，同时引入Confluent的Avro序列化依赖。
• 偏移量管理：若需要精确的消息处理（避免重复消费），可关闭enable.auto.commit，实现AcknowledgingMessageListener接口，手动提交偏移量。
• 并发配置：concurrency参数设置的消费线程数建议不超过目标Topic的分区数，否则会出现空闲线程。

内容的提问来源于stack exchange，提问作者Jordon