如何处理 Java8 并行流中的状态变量？

Question

我们正在使用 Java8 流中的操作链来处理事件流。作为处理的一部分，我们希望跟踪事件计数及其状态以进行测试和监控。这是我们的用例的简化示例，它为给定的日期流打印星期几。

public class StreamStateHandling {

   private static enum Status {RECEIVED, SUCCESS, ERROR};

   private Map<Status,Integer> results = new EnumMap<>(Status.class);

   private static final DateTimeFormatter formatter = DateTimeFormatter.ofPattern("MM/dd/yyyy");


   private static Optional<LocalDate> parseDate(String dateString){
    LocalDate localDate = null;
    try {
        localDate = LocalDate.from(formatter.parse(dateString));
    }catch (DateTimeParseException e){
        return Optional.empty();
    }
    return Optional.of(localDate);
   }

   private void doWork(){
    Stream.of("12/31/2014",
            "01-01-2015",
            "12/31/2015",
            "not a date",
            "01/01/2016")
            //.parallel()
            .peek(v -> addResult(Status.RECEIVED))
            .map(StreamStateHandling::parseDate)
            .peek(v -> {if (!v.isPresent()) addResult(Status.ERROR);})
            .filter(Optional::isPresent) 
            .map(Optional::get)
            .map(DayOfWeek::from) 
            .peek(v -> addResult(Status.SUCCESS))
            .forEach(System.out::println); 

       System.out.println(results);
  }
  public static void main(String args[]) {
    new StreamStateHandling().doWork();
  }

  private void addResult(Status status){

    int current = results.getOrDefault(status, 0);
    results.put(status, current + 1);
  }
}

基本上，我们在地图中跟踪状态计数。这在单线程处理中运行良好，但在并行流中会产生不确定的输出。

在现实世界中，我们有多种状态和操作链。一般来说，检测流和跟踪进度的最佳方法是什么？更喜欢 vanilla Java8 实现，但如果使用开源库更容易做到这一点，则可以。

非常感谢您的帮助。

Answer 1

EnumMap 不是线程安全的，addResult() 中的读-修改-写逻辑也不是。尝试使用原子ConcurrentHashMap.merge() 来增加计数：

private Map<Status, Integer> results = new ConcurrentHashMap<>();

private void addResult(Status status) {
    results.merge(status, 1, Integer::sum);
}