Java共享目录CSV文件更新内容异常隔离处理方案咨询

Great question! Since you already have a daily scheduled task runner up and running, you don’t need to switch to Java Watcher Service—there are several straightforward, effective alternatives to handle your requirement of processing CSV rows independently (even if one row fails). Let’s walk through them:

1. Enhance Your Existing Scheduled Task with Row-Level Error Handling #

This is the simplest approach since you can build directly on what you already have. The key is to:

• Track which rows you’ve already processed (e.g., store the last processed line number in a config file or database table)
• Wrap each row’s processing logic in its own try-catch block to isolate failures

Here’s a Java code snippet to illustrate this:

// Inside your scheduled task's execution logic
int lastProcessedLine = loadLastProcessedLine(); // Fetch from config/db
Path csvPath = Paths.get("path/to/your/shared/drive/csv");

try (BufferedReader reader = Files.newBufferedReader(csvPath)) {
    String line;
    int currentLine = 0;
    while ((line = reader.readLine()) != null) {
        currentLine++;
        if (currentLine <= lastProcessedLine) continue; // Skip already handled rows

        // Process each row independently
        try {
            if (currentLine == lastProcessedLine + 1) {
                saveToDatabase(line); // First new row: persist to DB
            } else {
                processSecondRow(line); // Second new row: run custom processing
            }
            updateLastProcessedLine(currentLine); // Mark row as processed (adjust if you want to mark even on failure)
        } catch (Exception e) {
            log.error("Failed to process line {}: {}", currentLine, e.getMessage());
            // Optional: Log the failed row to a retry queue or error log for later follow-up
            continue; // Move to next row regardless of failure
        }
    }
} catch (IOException e) {
    log.error("Error reading CSV file", e);
}

This ensures that even if the first row throws an exception, the second row’s processing will still execute without interruption.

2. Use a Message Queue for Decoupled, Independent Processing #

If you want to decouple the CSV reading step from the processing logic (and add retry capabilities for failed rows), a message queue is a great fit. Here’s how it works:

• Your scheduled task reads the new CSV rows and sends each one as a separate message to a queue (e.g., RabbitMQ, ActiveMQ, or even a lightweight Redis List)
• Separate consumer processes pick up each message and handle the row-specific logic (save to DB or custom processing)

Since each row is a standalone message, a failure in processing one message won’t block others. Failed messages can be requeued for later retries or routed to a dead-letter queue for manual review.

Example outline:

// Scheduled task: Read new rows and send to queue
List<String> newRows = fetchNewCsvRows(); // Implement to get only unprocessed rows
MessageQueue queue = getQueueClient(); // Initialize your queue client

for (String row : newRows) {
    // Add metadata if needed to identify first/second row, or infer from content
    queue.sendMessage(row);
}

// Consumer logic (runs independently)
while (true) {
    String row = queue.receiveMessage();
    try {
        if (isFirstRow(row)) { // Implement logic to detect row type
            saveToDatabase(row);
        } else {
            processSecondRow(row);
        }
        queue.acknowledgeMessage(); // Mark message as processed
    } catch (Exception e) {
        log.error("Failed processing row, requeuing", e);
        queue.requeueMessage(row); // Or send to dead-letter queue after retries
    }
}

3. Split Processing into Async Sub-Tasks (Framework-Based) #

If you’re using a framework like Spring, you can leverage asynchronous tasks to run each row’s processing in isolation. Your scheduled task triggers separate async sub-tasks for each row, ensuring that one failure doesn’t impact the other.

Example with Spring’s @Async:

// Scheduled task
@Scheduled(cron = "your-daily-cron-expression")
public void triggerCsvProcessing() {
    List<String> newRows = fetchNewCsvRows();
    
    if (!newRows.isEmpty()) {
        asyncTaskExecutor.execute(() -> processFirstRow(newRows.get(0)));
    }
    if (newRows.size() >= 2) {
        asyncTaskExecutor.execute(() -> processSecondRow(newRows.get(1)));
    }
}

// Isolated processing methods with error handling
private void processFirstRow(String row) {
    try {
        saveToDatabase(row);
    } catch (Exception e) {
        log.error("First row processing failed", e);
    }
}

private void processSecondRow(String row) {
    try {
        runCustomProcessing(row);
    } catch (Exception e) {
        log.error("Second row processing failed", e);
    }
}

Just make sure your async executor is configured properly to handle task isolation.

Final Recommendation #

Which option to choose depends on your needs:

• Go with row-level error handling if you want minimal changes to your existing setup.
• Use a message queue if you need decoupling, retry logic, or scalability.
• Pick async sub-tasks if you’re already using a framework that supports it (like Spring) and want clean, maintainable code.

内容的提问来源于stack exchange，提问作者Jyoti Prasad Dutta