Java Parallel Streams 关闭线程答案

【问题标题】：Java Parallel Streams close threadJava Parallel Streams 关闭线程
【发布时间】：2017-02-28 23:36:42
【问题描述】：

我已经编写了一个使用 Java 流的方法，它简单地迭代对象列表并返回 true/false 是否满足特定条件

Java 方法：

 boolean method(SampleObj sampleObj) {

   List testList = invokeSomeMethod();
   int result = testList
            .parallelStream()
            .filter(listObj -> (listObj.getAttr() = 1))
            .count(listObj -> listObj.isAttr4());

   return (result > 10);

 }

我也为此编写了一个 Mock 测试用例。当我执行测试用例时，测试成功，但是我收到项目自定义错误，指出创建的所有线程都没有关闭。

我什至尝试将流与 try-with-resources 一起使用，但 noo 没有帮助。

模拟测试：

@Test
public void testSomeMethod() {
    SampleObj sampleObj1 = new SampleObj(10, 20, 30, true);
    SampleObj sampleObj2 = new SampleObj(10, 20, 30, true);
    SampleObj sampleObj3 = new SampleObj(10, 20, 30, false);
    SampleObj sampleObjTest = new SampleObj(10, 20, 30, true);

    List<SampleObj> testList = new ArrayList<SampleObj>();
    testList.add(sampleObj1);
    testList.add(sampleObj2);
    testList.add(sampleObj3);

    when(mockedAttribute.invokeSomeMethod()).thenReturn(nodeList);

    ClassToBeTested classTest = createGenericMockRules();
    Assert.assertTrue(classTest.method(sampleObjTest));
}

附：我已经调试确认当调用 invokeSomeMethod() 时，我的模拟 testList 被返回。

据我所知，Java 流会在内部关闭它创建的线程。我是否执行不正确？

【问题讨论】：

你在嘲笑什么？你能展示你的测试用例吗？
添加代码sn-p
附带说明，== true 表达式总是多余的。
@learningMyWayThru 你能显示你得到的实际错误吗？
1.您正在使用原始类型 List。如果没有元素类型，则既不能找到 getAttr() 也不能找到 isAttr4()。 2. listObj.getAttr() = 1 是一个尝试的无效assignment。 3.Stream.count()没有参数。为什么要发布该代码，而它显然与真实代码无关？

标签： java multithreading unit-testing java-stream

【解决方案1】：

Java 流不创建线程，因此不释放线程。他们在内部使用线程池；未指定，但众所周知它是common pool of the Fork/Join framework。

使用线程池的全部目的是让池来管理线程，而不是为每个作业创建和释放线程。创建和销毁线程与成本相关，当随后将多个作业排队时应避免这种情况。特别是如果没有现有的线程来接它，线程的创建时间会增加作业的执行时间。换句话说，线程比作业寿命更长是正常的并且是有意的。他们正在等待可能到来的新工作。

class documentation of ForkJoinPool 声明：

静态commonPool() 适用于大多数应用程序。任何未明确提交到指定池的 ForkJoinTask 都使用公共池。使用公共池通常会减少资源使用（其线程在不使用期间会缓慢回收，并在后续使用时恢复）。

除了“缓慢”之外，它没有指定线程在被回收之前必须空闲的时间，因此它甚至可能因实现而异。对于当前的实现，甚至不能说超时，因为池会收缩线程数，而不是在超时后终止所有空闲线程，因此剩余线程将再次等待，增加超时，直到池再次缩小大小，直到没有空闲线程。换句话说，池中的线程越多，直到最后一个线程被回收的时间越长，此时所有线程都处于空闲状态。

您可以通过以下方式强制测试等待所有线程结束

while(ForkJoinPool.commonPool().getPoolSize()>0)
    LockSupport.parkNanos(1000);

但这会显着增加测试的执行时间，例如谈论八核/线程一分钟的大小。更好的解决方案是重新考虑您的“项目自定义错误”检查，即根本不应该认为您的代码对内部使用的池创建的线程负责。

否则，您在使用时可能会遇到类似的错误，例如异步 I/O 等

【讨论】：

非常棒的答案！