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

Question

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

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);
  }
}

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

在现实世界中，我们有几个状态和操作链。总而言之，检测流和跟踪进度的最佳方法是什么？我更喜欢普通的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);
}