如何在目录新增文件时自动触发Scala编写的Kafka生产者与消费者？

Hey there! Let's tackle this problem of automating your Kafka producer/consumer workflow when new files drop into your target directory. Here are a few solid approaches tailored to your Scala/Kafka setup:

If you prefer keeping everything within your Scala codebase without external tools, JDK's built-in WatchService is perfect for real-time directory monitoring. It lets you listen for file creation events and trigger your producer logic automatically.

Key Steps: #

• Integrate WatchService into your producer to monitor the target directory for ENTRY_CREATE events (new files).
• Keep your consumer running as a long-lived process (Kafka consumers are designed for this!) instead of triggering it per file—this avoids redundant startup overhead and maintains proper offset tracking.

Example Scala Code Snippet: #

import java.nio.file._
import java.nio.file.attribute.BasicFileAttributes

object DirectoryWatcher {
  def watchDirectory(path: String, onFileAdded: Path => Unit): Unit = {
    val dir = Paths.get(path)
    val watchService = FileSystems.getDefault.newWatchService()
    dir.register(watchService, StandardWatchEventKinds.ENTRY_CREATE)

    println(s"Watching directory: $path for new files...")
    while (true) {
      val key = watchService.take()
      for (event <- key.pollEvents()) {
        val kind = event.kind()
        if (kind == StandardWatchEventKinds.ENTRY_CREATE) {
          val filePath = dir.resolve(event.context().asInstanceOf[Path])
          // Skip directories, only process regular files
          if (Files.isRegularFile(filePath)) {
            onFileAdded(filePath)
          }
        }
      }
      key.reset()
    }
  }
}

// Your main producer/consumer orchestration
object KafkaFileWorkflow {
  def main(args: Array[String]): Unit = {
    val targetDir = "/path/to/your/target/directory"
    
    // Start consumer as a background thread (long-lived)
    startKafkaConsumer()

    // Start directory watcher to trigger producer on new files
    DirectoryWatcher.watchDirectory(targetDir, filePath => {
      println(s"New file detected: $filePath")
      readFileAndSendToKafka(filePath)
    })
  }

  def startKafkaConsumer(): Unit = {
    new Thread(() => {
      // Your existing Kafka consumer logic here:
      // - Create KafkaConsumer instance
      // - Subscribe to your target topic
      // - Loop to poll and process messages
    }).start()
  }

  def readFileAndSendToKafka(filePath: Path): Unit = {
    // Your existing file-reading and Kafka producer logic here
  }
}

Pros & Cons: #

• ✅ Cross-platform (works on Windows, macOS, Linux)
• ✅ Full control within your codebase
• ❌ Requires modifying existing code to integrate the watcher logic

If you don't want to touch your Scala code, use Linux's inotify-tools to monitor the directory and trigger your existing producer script. This is great for quick automation without code modifications.

Key Steps: #

1. Install inotify-tools (Linux-only):

   # Debian/Ubuntu
   sudo apt-get install inotify-tools
   
   # CentOS/RHEL
   sudo yum install inotify-tools
   

2. Write a shell script to start your consumer as a background service and watch for new files.

Example Shell Script (auto_kafka_workflow.sh): #

#!/bin/bash

TARGET_DIR="/path/to/your/target/directory"
PRODUCER_JAR="/path/to/your/compiled-producer.jar"
CONSUMER_JAR="/path/to/your/compiled-consumer.jar"

# Start consumer in background (persistent process)
echo "Starting Kafka consumer in background..."
nohup java -jar $CONSUMER_JAR > consumer.log 2>&1 &

# Monitor directory for new files
echo "Watching directory $TARGET_DIR..."
inotifywait -m -e create --format '%w%f' $TARGET_DIR | while read NEW_FILE
do
  # Only process regular files (skip subdirectories)
  if [ -f "$NEW_FILE" ]; then
    echo "Processing new file: $NEW_FILE"
    # Run your producer, passing the new file path as an argument
    java -jar $PRODUCER_JAR "$NEW_FILE"
  fi
done

3. Make the script executable and run it:

   chmod +x auto_kafka_workflow.sh
   ./auto_kafka_workflow.sh
   

Pros & Cons: #

• ✅ No changes needed to your Scala code
• ✅ Lightweight and easy to set up
• ❌ Linux-only (won't work on Windows/macOS without extra tools like WSL)

If you anticipate scaling to more complex data pipelines (e.g., file validation, error retries, monitoring), Apache NiFi is a great low-code option:

• Use NiFi's GetFile processor to watch your target directory for new files
• Use PublishKafkaRecord to send file content to your Kafka topic
• Use ConsumeKafkaRecord to process messages downstream
• NiFi provides built-in monitoring, retry logic, and visual pipeline design

This is overkill for simple automation, but perfect if your workflow will grow in complexity.

Quick Tips: #

• Consumer Best Practice: Always run your Kafka consumer as a persistent process instead of triggering it per file—this maintains proper offset tracking and avoids redundant startup.
• Idempotency: Add logic to track processed files (e.g., a database table or a "processed" directory) to avoid re-sending duplicate messages if files are re-added.
• Error Handling: If your producer fails to send a file, move the file to an error directory instead of letting the script loop indefinitely.

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