如何用SpringBoot从文件实时向Kafka Topic发送数据？

基于你已有的Kafka发送逻辑，以下是实现从文件实时读取数据并发送到Kafka的具体方案：

1. 添加依赖 #

如果使用Maven，在pom.xml中加入Apache Commons IO依赖（用于实时监控文件新增内容）：

<dependency>
    <groupId>commons-io</groupId>
    <artifactId>commons-io</artifactId>
    <version>2.15.1</version>
</dependency>

2. 实现文件实时监控与发送服务 #

创建一个服务类，负责监控文件新增内容、解析为Order对象并调用已有的KafkaProducerService发送：

import org.apache.commons.io.input.Tailer;
import org.apache.commons.io.input.TailerListenerAdapter;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import java.io.File;

@Service
public class FileKafkaSenderService {

    @Autowired
    private KafkaProducerService kafkaProducerService;

    @Autowired
    private ObjectMapper objectMapper;

    // 从配置文件读取监控路径，更灵活
    @Value("${kafka.file.monitor.path}")
    private String monitorFilePath;

    @PostConstruct
    public void startFileMonitoring() {
        File targetFile = new File(monitorFilePath);
        // 每秒检查一次文件是否有新增内容
        Tailer fileTailer = new Tailer(targetFile, new TailerListenerAdapter() {
            @Override
            public void handle(String line) {
                try {
                    // 假设文件每行是一个JSON格式的Order对象（JSON Lines格式）
                    Order order = objectMapper.readValue(line, Order.class);
                    kafkaProducerService.send(order);
                } catch (Exception e) {
                    // 可根据需求优化异常处理，比如记录到日志或死信队列
                    System.err.println("解析并发送文件行失败: " + line);
                    e.printStackTrace();
                }
            }
        }, 1000);

        // 启动监控线程
        new Thread(fileTailer).start();
    }
}

3. 配置文件路径 #

在application.yml或application.properties中配置要监控的文件路径：

kafka:
  file:
    monitor:
      path: /your/actual/file/path/orders.jsonl

4. 可选：一次性读取整个文件发送 #

如果不需要实时监控新增内容，而是启动时一次性读取整个文件并发送，可替换startFileMonitoring方法为：

import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.List;

@PostConstruct
public void sendFileDataOnStartup() {
    try {
        List<String> lines = Files.readAllLines(Paths.get(monitorFilePath));
        for (String line : lines) {
            Order order = objectMapper.readValue(line, Order.class);
            kafkaProducerService.send(order);
        }
    } catch (Exception e) {
        System.err.println("读取文件并发送失败");
        e.printStackTrace();
    }
}

注意事项 #

• 文件格式：上述代码假设文件采用JSON Lines格式（每行一个独立的JSON对象），如果是其他格式（比如整个文件是JSON数组），需要调整解析逻辑。
• 线程管理：示例中用了基础线程启动，生产环境建议使用Spring的TaskExecutor来统一管理线程池，避免线程混乱。
• 异常处理：可根据业务需求扩展，比如将解析失败的内容写入单独的日志文件或死信Topic。

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