MK
摩柯社区 - 一个极简的技术知识社区
AI 面试

Java 流同步与异步的错误处理

2023-04-136.8k 阅读

Java 流同步与异步的错误处理

在Java开发中，流（Stream）是处理数据序列的强大工具，无论是同步还是异步操作，都可能会遇到各种错误。妥善处理这些错误对于确保程序的健壮性和稳定性至关重要。接下来，我们将深入探讨Java流同步与异步场景下的错误处理机制。

Java 同步流的错误处理

1. 常见错误类型

  • 空指针异常（NullPointerException）：当流操作涉及到空值时，可能会抛出此异常。例如，在对一个可能为空的集合创建流并进行操作时，如果没有提前检查，就容易出现问题。
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

public class SyncStreamErrorHandling {
    public static void main(String[] args) {
        List<String> list = null;
        try {
            List<Integer> lengths = list.stream()
                  .map(String::length)
                  .collect(Collectors.toList());
        } catch (NullPointerException e) {
            System.out.println("捕获到空指针异常: " + e.getMessage());
        }
    }
}

在上述代码中，list 被初始化为 null，当尝试对其创建流并进行操作时，就会抛出 NullPointerException。通过 try - catch 块，我们可以捕获并处理这个异常。

  • 类型转换异常（ClassCastException）：如果流操作中进行了不恰当的类型转换，就会出现此异常。比如，在一个混合类型的集合中，错误地将非目标类型转换为目标类型。
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

public class SyncStreamErrorHandling {
    public static void main(String[] args) {
        List<Object> mixedList = new ArrayList<>();
        mixedList.add("1");
        mixedList.add(2);
        try {
            List<Integer> intList = mixedList.stream()
                  .map(s -> (Integer) s)
                  .collect(Collectors.toList());
        } catch (ClassCastException e) {
            System.out.println("捕获到类型转换异常: " + e.getMessage());
        }
    }
}

这里，mixedList 包含了字符串和整数，当尝试将所有元素转换为 Integer 时，就会对字符串元素抛出 ClassCastException。

2. 使用 Optional 处理潜在空值

Optional 类是Java 8引入的用于处理可能为空的值的工具。在流操作中，可以利用它来避免空指针异常。

import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;

public class SyncStreamErrorHandling {
    public static void main(String[] args) {
        List<String> list = null;
        List<Integer> lengths = Optional.ofNullable(list)
              .orElseGet(ArrayList::new)
              .stream()
              .mapToInt(String::length)
              .boxed()
              .collect(Collectors.toList());
        System.out.println(lengths);
    }
}

在这段代码中，Optional.ofNullable(list) 首先检查 list 是否为空，如果为空，则通过 orElseGet 方法提供一个空的 ArrayList。这样，在后续的流操作中就不会因为 list 为空而抛出 NullPointerException。

3. 自定义错误处理

有时候，标准的异常类型不能满足业务需求，需要自定义异常。例如，在一个处理用户输入的流操作中，如果输入不符合特定格式，可以抛出自定义异常。

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

class InvalidInputException extends Exception {
    public InvalidInputException(String message) {
        super(message);
    }
}

public class SyncStreamErrorHandling {
    public static void main(String[] args) {
        List<String> inputs = new ArrayList<>();
        inputs.add("abc");
        inputs.add("123");
        try {
            List<Integer> validNumbers = inputs.stream()
                  .map(SyncStreamErrorHandling::parseInput)
                  .collect(Collectors.toList());
            System.out.println(validNumbers);
        } catch (InvalidInputException e) {
            System.out.println("捕获到自定义异常: " + e.getMessage());
        }
    }

    private static Integer parseInput(String input) throws InvalidInputException {
        try {
            return Integer.parseInt(input);
        } catch (NumberFormatException e) {
            throw new InvalidInputException("无效输入: " + input);
        }
    }
}

在上述代码中，parseInput 方法尝试将输入字符串解析为整数，如果解析失败，就抛出 InvalidInputException 自定义异常。在流操作的 map 方法中调用这个方法，并在外部使用 try - catch 块捕获自定义异常。

Java 异步流的错误处理

1. 异步流基础

异步流允许在不阻塞主线程的情况下处理数据。Java 9引入了 Flow API来支持异步流。Flow 包含四个主要接口：Publisher（发布者）、Subscriber（订阅者）、Subscription（订阅关系）和 Processor（处理者，是 Publisher 和 Subscriber 的组合）。

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

public class AsyncStreamExample {
    public static void main(String[] args) {
        SubmissionPublisher<Integer> publisher = new SubmissionPublisher<>();
        Flow.Subscriber<Integer> subscriber = new Flow.Subscriber<>() {
            private Flow.Subscription subscription;

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                this.subscription = subscription;
                subscription.request(1);
            }

            @Override
            public void onNext(Integer item) {
                System.out.println("接收到数据: " + item);
                subscription.request(1);
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println("捕获到错误: " + throwable.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println("数据处理完成");
            }
        };
        publisher.subscribe(subscriber);
        for (int i = 0; i < 5; i++) {
            publisher.submit(i);
        }
        publisher.close();
    }
}

在这个示例中，SubmissionPublisher 作为发布者，Flow.Subscriber 作为订阅者。订阅者通过 onSubscribe 方法请求数据，在 onNext 方法中处理接收到的数据，onError 方法处理错误，onComplete 方法表示数据处理完成。

2. 异步流中的错误处理

  • 错误传播：在异步流中，错误需要正确地传播给订阅者。发布者可以通过 Subscription 的 cancel 方法来终止订阅关系，并在 onError 方法中通知订阅者错误。
import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

public class AsyncStreamErrorHandling {
    public static void main(String[] args) {
        SubmissionPublisher<Integer> publisher = new SubmissionPublisher<>();
        Flow.Subscriber<Integer> subscriber = new Flow.Subscriber<>() {
            private Flow.Subscription subscription;

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                this.subscription = subscription;
                subscription.request(1);
            }

            @Override
            public void onNext(Integer item) {
                if (item == 3) {
                    subscription.cancel();
                    throw new RuntimeException("遇到错误数据");
                }
                System.out.println("接收到数据: " + item);
                subscription.request(1);
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println("捕获到错误: " + throwable.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println("数据处理完成");
            }
        };
        publisher.subscribe(subscriber);
        for (int i = 0; i < 5; i++) {
            publisher.submit(i);
        }
        publisher.close();
    }
}

在上述代码中，当接收到数据 3 时，订阅者通过 subscription.cancel() 终止订阅，并抛出 RuntimeException。发布者会捕获这个异常，并通过 onError 方法通知其他订阅者。

  • 使用 CompletableFuture 处理异步流错误：CompletableFuture 可以方便地处理异步操作的结果和错误。在异步流中，可以将流操作封装在 CompletableFuture 中，并使用 exceptionally 方法处理错误。
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class AsyncStreamErrorHandling {
    public static void main(String[] args) {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            try {
                return Stream.of("1", "2", "abc", "4")
                      .map(Integer::parseInt)
                      .map(Object::toString)
                      .collect(Collectors.joining(","));
            } catch (NumberFormatException e) {
                throw new RuntimeException("解析错误", e);
            }
        });
        future.exceptionally(ex -> {
            System.out.println("捕获到错误: " + ex.getMessage());
            return "默认值";
        }).thenAccept(System.out::println);
    }
}

这里，CompletableFuture.supplyAsync 方法异步执行流操作。如果在流操作中抛出 NumberFormatException，exceptionally 方法会捕获这个异常，并返回一个默认值。

3. 异步流错误处理的最佳实践

  • 集中错误处理：在复杂的异步流场景中，建议采用集中式的错误处理机制。可以创建一个全局的错误处理器，对所有异步流操作中的错误进行统一处理。
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;
import java.util.stream.Stream;

class GlobalErrorHandler {
    public static void handleError(Throwable throwable) {
        System.out.println("全局错误处理: " + throwable.getMessage());
    }
}

public class AsyncStreamBestPractice {
    private static final ExecutorService executor = Executors.newFixedThreadPool(5);

    public static void main(String[] args) {
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            try {
                return Stream.of("1", "2", "abc", "4")
                      .map(Integer::parseInt)
                      .map(Object::toString)
                      .collect(Collectors.joining(","));
            } catch (NumberFormatException e) {
                throw new RuntimeException("解析错误", e);
            }
        }, executor);
        future1.exceptionally(ex -> {
            GlobalErrorHandler.handleError(ex);
            return "默认值1";
        }).thenAccept(System.out::println);

        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            try {
                return Stream.of("5", "6", "7")
                      .map(Integer::parseInt)
                      .map(Object::toString)
                      .collect(Collectors.joining(","));
            } catch (Exception e) {
                throw new RuntimeException("其他错误", e);
            }
        }, executor);
        future2.exceptionally(ex -> {
            GlobalErrorHandler.handleError(ex);
            return "默认值2";
        }).thenAccept(System.out::println);

        executor.shutdown();
    }
}

在这个示例中，GlobalErrorHandler 类提供了一个静态方法 handleError 用于集中处理错误。多个 CompletableFuture 操作中的错误都通过这个全局处理器进行处理。

  • 日志记录：在处理异步流错误时，详细的日志记录是非常重要的。通过日志可以追踪错误发生的时间、位置以及相关的上下文信息，有助于快速定位和解决问题。
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class AsyncStreamErrorLogging {
    private static final Logger logger = Logger.getLogger(AsyncStreamErrorLogging.class.getName());

    public static void main(String[] args) {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            try {
                return Stream.of("1", "2", "abc", "4")
                      .map(Integer::parseInt)
                      .map(Object::toString)
                      .collect(Collectors.joining(","));
            } catch (NumberFormatException e) {
                logger.log(Level.SEVERE, "解析错误", e);
                throw new RuntimeException("解析错误", e);
            }
        });
        future.exceptionally(ex -> {
            logger.log(Level.SEVERE, "捕获到异常", ex);
            return "默认值";
        }).thenAccept(System.out::println);
    }
}

在上述代码中，Logger 类用于记录错误信息。log 方法的第一个参数指定日志级别，第二个参数是错误信息，第三个参数是异常对象。通过这种方式，可以详细记录异步流操作中的错误情况。

同步与异步流错误处理的对比

  1. 错误捕获方式
    • 同步流：主要通过传统的 try - catch 块来捕获错误。这种方式简单直接，在流操作的同一线程中处理错误，容易理解和调试。例如，在同步流操作涉及文件读取时，如果文件不存在，通过 try - catch 捕获 FileNotFoundException 并进行处理。
import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class SyncStreamFileRead {
    public static void main(String[] args) {
        try (BufferedReader br = new BufferedReader(new FileReader("nonexistent.txt"))) {
            String content = br.lines()
                  .collect(Collectors.joining("\n"));
            System.out.println(content);
        } catch (IOException e) {
            System.out.println("文件读取错误: " + e.getMessage());
        }
    }
}

- **异步流**：异步流的错误捕获更加依赖于 `onError` 回调方法（如 `Flow.Subscriber` 中的 `onError`）或者 `CompletableFuture` 的 `exceptionally` 方法。由于异步操作在不同线程执行，错误处理需要通过回调机制来实现。例如，在异步读取文件内容并处理时，通过 `CompletableFuture` 来处理可能的错误。

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class AsyncStreamFileRead {
    public static void main(String[] args) {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            try (BufferedReader br = new BufferedReader(new FileReader("nonexistent.txt"))) {
                return br.lines()
                      .collect(Collectors.joining("\n"));
            } catch (IOException e) {
                throw new RuntimeException("文件读取错误", e);
            }
        });
        future.exceptionally(ex -> {
            System.out.println("捕获到错误: " + ex.getMessage());
            return "默认内容";
        }).thenAccept(System.out::println);
    }
}
  1. 错误传播与处理的复杂度
    • 同步流：错误传播相对简单，因为所有操作都在同一线程执行。如果一个流操作抛出异常，它会直接中断当前流的处理，并且异常会按照调用栈向上传播，直到被 try - catch 块捕获。例如，在一个链式的同步流操作中，map 方法抛出异常，后续的 filter 和 collect 方法不会被执行，异常会被最近的 try - catch 块捕获。
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

public class SyncStreamErrorPropagation {
    public static void main(String[] args) {
        List<String> list = new ArrayList<>();
        list.add("1");
        list.add("abc");
        try {
            List<Integer> intList = list.stream()
                  .map(Integer::parseInt)
                  .filter(i -> i > 0)
                  .collect(Collectors.toList());
        } catch (NumberFormatException e) {
            System.out.println("捕获到异常: " + e.getMessage());
        }
    }
}

- **异步流**：错误传播较为复杂，因为涉及多个线程和异步操作。发布者需要通过订阅关系来传播错误给订阅者，并且不同的异步框架（如 `Flow` API、`CompletableFuture` 等）有不同的错误传播机制。例如，在 `Flow` API中，发布者通过 `Subscription` 的 `cancel` 方法和 `onError` 回调通知订阅者错误；而在 `CompletableFuture` 中，错误通过 `exceptionally` 方法处理，并且可以通过 `thenApply` 等方法将错误传递给后续的异步操作。

import java.util.concurrent.CompletableFuture;

public class AsyncStreamErrorPropagation {
    public static void main(String[] args) {
        CompletableFuture.supplyAsync(() -> {
            if (Math.random() > 0.5) {
                throw new RuntimeException("随机错误");
            }
            return "数据";
        })
              .thenApply(String::toUpperCase)
              .exceptionally(ex -> {
                    System.out.println("捕获到异常: " + ex.getMessage());
                    return "默认值";
                })
              .thenAccept(System.out::println);
    }
}
  1. 资源管理与错误处理的结合
    • 同步流：在同步流操作中，资源管理（如文件关闭、数据库连接关闭等）通常与错误处理紧密结合。可以使用 try - with - resources 语句确保资源在操作完成后正确关闭，即使发生异常也能保证资源的释放。例如，在读取文件的同步流操作中，try - with - resources 会自动关闭 BufferedReader。
import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class SyncStreamResourceManagement {
    public static void main(String[] args) {
        try (BufferedReader br = new BufferedReader(new FileReader("example.txt"))) {
            String content = br.lines()
                  .collect(Collectors.joining("\n"));
            System.out.println(content);
        } catch (IOException e) {
            System.out.println("文件读取错误: " + e.getMessage());
        }
    }
}

- **异步流**：异步流的资源管理相对复杂，因为异步操作可能在不同线程执行，资源的生命周期管理需要更加小心。例如，在异步读取文件内容时，如果使用自定义的线程池来执行异步操作，需要确保在操作完成或发生错误时，正确关闭文件资源和线程池。一种常见的做法是在 `CompletableFuture` 的 `whenComplete` 方法中处理资源关闭。

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class AsyncStreamResourceManagement {
    private static final ExecutorService executor = Executors.newFixedThreadPool(5);

    public static void main(String[] args) {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            BufferedReader br = null;
            try {
                br = new BufferedReader(new FileReader("example.txt"));
                return br.lines()
                      .collect(Collectors.joining("\n"));
            } catch (IOException e) {
                throw new RuntimeException("文件读取错误", e);
            } finally {
                if (br != null) {
                    try {
                        br.close();
                    } catch (IOException e) {
                        System.out.println("文件关闭错误: " + e.getMessage());
                    }
                }
            }
        }, executor);
        future.whenComplete((result, ex) -> {
            executor.shutdown();
            if (ex != null) {
                System.out.println("捕获到错误: " + ex.getMessage());
            } else {
                System.out.println("结果: " + result);
            }
        });
    }
}

总结与最佳实践回顾

在Java流的开发中，无论是同步还是异步流，错误处理都是保证程序健壮性的关键环节。对于同步流，要善于利用 try - catch 块、Optional 类以及自定义异常来处理常见错误，如空指针异常、类型转换异常等。在异步流方面，要深入理解 Flow API 和 CompletableFuture 的错误处理机制，通过 onError 回调、exceptionally 方法等进行错误捕获和处理。

同时，无论是同步还是异步流，都有一些通用的最佳实践。例如，集中错误处理可以提高代码的可维护性，将所有流操作的错误统一交给一个全局的错误处理器处理；详细的日志记录能够帮助快速定位和解决问题，在错误发生时记录关键的上下文信息。此外，合理的资源管理与错误处理相结合也是必不可少的，确保在发生错误时资源能够正确释放，避免资源泄漏。通过遵循这些原则和最佳实践，开发者能够编写出更加健壮、可靠的Java流应用程序。