Java 8 Stream：迭代、处理和计数

Question

在这种方式下处理和计算已处理的数据是否可行？

long count = userDao.findApprovedWithoutData().parallelStream().filter(u -> {
    Data d = dataDao.findInfoByEmail(u.getEmail());
    boolean ret = false;
    if (d != null) {
        String result = "";
        result += getFieldValue(d::getName, ". \n");
        result += getFieldValue(d::getOrganization, ". \n");
        result += getFieldValue(d::getAddress, ". \n");
        if(!result.isEmpty()) {
            u.setData(d.getInfo());
            userDao.update(u);
            ret = true;
        }
    }
    return ret;
}).count();

因此，简而言之：迭代不完整的记录，如果存在数据则更新并计算这些记录的数量？

- user_x

---

你最好在http://programmers.stackexchange.com/上询问这个问题。 - Philipp Sander

userDao.update 是做什么的？它是线程安全的吗？ - Louis Wasserman

@PhilippSander 当提及其他网站时，指出不赞成跨贴通常是有帮助的。 - gnat

@Louis Wasserman 更新数据库。线程安全 :) - user_x

Answer 1

我认为这是糟糕的代码，因为：

筛选谓词具有（相当显著的）副作用

谓词不应该具有副作用（就像getter一样）。这是出乎意料的，这使得它变得糟糕。

筛选谓词非常低效

每次执行谓词都会导致大量查询链的触发，这使得这段代码不可扩展。

乍一看，主要目的似乎是获取计数，但实际上这只是一个次要（可有可无）的信息

好的代码使得正在进行的操作显而易见（不像这段代码）

您应该更改代码，使用一个（相当简单的）单个更新查询（利用连接），并从持久性API结果中的“更新行数”信息获取计数。

- Bohemian

---

如何正确编写此代码的策略/构思（不查看数据库以计算结果）？ - user_x

Answer 2

这取决于您对process的定义。我无法给出明确的yes or no，因为我认为在不了解您的代码及其实现方式的情况下很难得出结论。

您正在使用Parallel Stream，Java运行时会根据ForkJoinPool的通用池中可用线程的数量将Stream拆分为子流。

使用parallelism时需要注意可能出现的副作用：

1. 干扰（流中的Lambda表达式不应该干扰）

流操作中的Lambda表达式不应干扰。 当流的源在流水线处理流时被修改时，就会发生干扰。

2. 有状态的Lambda表达式

避免在流操作的参数中使用有状态的Lambda表达式。有状态的Lambda表达式是指其结果取决于在执行流程中可能发生变化的任何状态。

看着你的问题并将上述要点应用于它。

非干扰性 > 强烈声明Lambda表达式在流的管道操作期间不应干扰流的源（除非流源是并发的），因为这可能会导致：

• 异常（即ConcurrentModificationException）
• 错误答案
• 不符合行为

除了良好行为的流，在中间操作（即过滤器）期间进行修改的情况下，阅读更多 here。

您的Lambda表达式确实干扰了流的源，这是不建议的，但是干扰发生在Intermediate操作中，现在一切都取决于流是否具有良好的行为。因此，当涉及到干扰时，您可能需要重新考虑Lambda表达式。这也可能取决于您如何通过userDao.udpate更新流的源，这从您的问题中并不清楚。

有状态 Lambda 表达式 > 您的 Lambda 表达式似乎并非具有状态，因为 Lambda 的结果取决于在管道执行期间不会更改的一个或多个值。因此，这并不适用于您的情况。

我建议您查阅Java 8 Stream文档以及这个博客，其中提供了很好的Java 8 Stream示例和解释。

- Raf