如何消费Apache Kafka消息并解析生成XML文件后发送至HTTPS网关

一、现成轻量级替代方案（无需从零编写Java代码） #

• 优先选择Apache Camel：内置Kafka消费者、XML格式转换、HTTP调用组件，不需要编写复杂业务逻辑，仅配置路由规则即可实现全流程，打包后是单jar包可直接运行，资源占用极低，完全符合轻量要求。
• 可选Spring Cloud Stream：如果后续有扩展微服务架构的需求，绑定Kafka binder后，仅需编写消息处理的核心逻辑，消费重试、偏移量管理、异常兜底等能力都已封装完成。
• 零代码方案可选择NiFi：通过拖拽配置即可完成「拉取Kafka消息→转换为XML格式→POST到HTTPS接口」的全流程，缺点是资源占用比前两个方案高，单机运行无压力。

二、Java应用开发入门步骤（最简实现，适配Java基础薄弱的场景） #

前置依赖 #

用Maven管理依赖，仅需引入三个核心依赖即可：

<!-- Kafka官方客户端 -->
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.4.0</version>
</dependency>
<!-- JAXB 用于生成XML，JDK8及以下版本自带无需引入，高版本JDK需单独添加 -->
<dependency>
    <groupId>com.sun.xml.bind</groupId>
    <artifactId>jaxb-impl</artifactId>
    <version>2.3.3</version>
    <scope>runtime</scope>
</dependency>
<!-- OkHttp 用于发送HTTPS POST请求 -->
<dependency>
    <groupId>com.squareup.okhttp3</groupId>
    <artifactId>okhttp</artifactId>
    <version>4.10.0</version>
</dependency>

核心代码实现 #

1. 初始化Kafka消费者

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "你的Kafka服务地址:端口");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "自定义消费组ID");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
// 简单场景开启自动提交偏移量即可
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
// 首次消费从最新消息开始
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("你的Kafka Topic名称"));

2. 循环消费+XML生成+接口调用

先定义和目标XML结构对应的Java实体类，添加JAXB注解，收到Kafka消息后把字段映射到实体类，再生成XML发送即可：

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        // 1. 解析Kafka消息内容，映射到XML对应的实体类
        YourXmlEntity entity = parseMessageToEntity(record.value());
        // 2. 生成XML内容
        JAXBContext jaxbContext = JAXBContext.newInstance(YourXmlEntity.class);
        Marshaller marshaller = jaxbContext.createMarshaller();
        marshaller.setProperty(Marshaller.JAXB_FORMATTED_OUTPUT, true);
        String xmlContent;
        try(StringWriter writer = new StringWriter()) {
            marshaller.marshal(entity, writer);
            xmlContent = writer.toString();
        }
        // 3. POST发送到HTTPS网关
        OkHttpClient client = new OkHttpClient();
        RequestBody body = RequestBody.create(xmlContent, MediaType.get("application/xml; charset=utf-8"));
        Request request = new Request.Builder()
                .url("你的HTTPS网关地址")
                .post(body)
                .build();
        try (Response response = client.newCall(request).execute()) {
            if (!response.isSuccessful()) throw new IOException("请求失败: " + response);
            // 可自行添加成功日志打印逻辑
        }
    }
}

注意事项 #

• 简单场景不需要引入Spring全家桶，上述代码打包为可执行jar包即可直接运行，内存占用不到100M。
• 生产环境可添加简单的异常重试逻辑，消费失败的消息可写入死信Topic避免阻塞正常流程。
• 如果Kafka消息是JSON格式，添加jackson-databind依赖即可直接反序列化，不需要手动写解析逻辑。

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