Java多线程问题：文件未归档且Kafka消息重复

一、文件未归档的原因 #

• Kafka发送异常：kafkaProducer.send() 若抛出序列化错误、网络连接失败或集群不可用等异常，会触发catch块，导致后续文件移动代码无法执行。另外，默认send为异步操作，若未处理回调中的异常，可能出现消息发送失败但主线程无感知的情况，若为同步异常（如配置错误）则直接中断流程。
• 文件读取异常：WatchService的ENTRY_CREATE事件可能在文件尚未完全写入时触发（比如写入进程仍占用文件），此时调用Files.readString()会抛出IO异常，中断后续移动操作。
• 文件移动操作失败：
  • 归档目录不存在且未自动创建，抛出NoSuchFileException；
  • 程序无归档目录写入权限，抛出AccessDeniedException；
  • 归档目录已存在同名文件，默认Files.move()不覆盖，抛出FileAlreadyExistsException；
  • 文件被其他进程锁定，无法执行移动。
• 线程池任务中断：线程池中的线程被意外中断，导致移动代码未执行完毕。

二、确保文件内容仅发送一次到Kafka的解决建议 #

1. 先锁定文件避免重复处理 #

通过原子重命名操作抢占文件所有权，确保同一文件仅被一个线程处理：

threadPool.submit(() -> {
    Path tempPath = filePath.resolveSibling(filePath.getFileName() + ".processing");
    // 尝试原子重命名，成功则说明文件未被处理
    try {
        Files.move(filePath, tempPath, StandardCopyOption.ATOMIC_MOVE);
    } catch (FileAlreadyExistsException e) {
        return; // 重命名失败，说明已有线程在处理，直接跳过
    }
    try {
        String content = Files.readString(tempPath);
        // 同步发送并确认Kafka消息已提交
        kafkaProducer.send(new ProducerRecord<>("topic", content)).get();
        // 移动到归档目录
        Files.move(tempPath, Paths.get("archive_directory").resolve(filePath.getFileName()));
    } catch (Exception e) {
        e.printStackTrace();
        // 处理失败，将临时文件恢复原名称以便重试
        Files.move(tempPath, filePath, StandardCopyOption.ATOMIC_MOVE);
    }
});

2. 验证文件完整性 #

由于ENTRY_CREATE可能在文件写入过程中触发，需等待文件写入完成后再处理：

// 轮询检查文件大小是否稳定，确认写入完成
long prevSize = -1;
while (true) {
    long currentSize = Files.size(filePath);
    if (currentSize == prevSize) {
        break;
    }
    prevSize = currentSize;
    Thread.sleep(100); // 短暂等待，避免频繁轮询
}

3. 记录已处理文件标识 #

使用线程安全集合记录已处理（或正在处理）的文件唯一标识，避免重复提交：

private final ConcurrentHashMap<String, Boolean> processingFiles = new ConcurrentHashMap<>();

// 任务处理逻辑中：
String fileUniqueId = filePath.toAbsolutePath().toString();
// 原子性存入，若已存在则跳过
if (processingFiles.putIfAbsent(fileUniqueId, true) != null) {
    return;
}
try {
    // 读取、发送Kafka、移动文件的逻辑
    processingFiles.remove(fileUniqueId);
} catch (Exception e) {
    processingFiles.remove(fileUniqueId); // 处理失败移除标记，允许重试
    e.printStackTrace();
}

4. 确保Kafka发送与文件移动的原子性 #

使用Kafka同步发送（get()方法阻塞直到消息确认），确保消息发送成功后再执行文件移动；若发送失败，保留原文件以便后续重试：

// 替换原异步send调用，同步等待发送结果
RecordMetadata metadata = kafkaProducer.send(new ProducerRecord<>("topic", content)).get();
// 确认发送成功后再移动文件
Files.move(filePath, Paths.get("archive_directory").resolve(filePath.getFileName()));

5. 优化WatchService事件处理 #

部分系统（如Windows）会重复触发ENTRY_CREATE事件，可在处理前检查文件是否已归档，若已存在则直接跳过当前事件。

内容的提问来源于stack exchange，提问作者surrya U R