Java 8 stream.map() :线程异常 "main" java.lang.IllegalStateException:源已经消耗或关闭
Java 8 stream.map() : Exception in thread "main" java.lang.IllegalStateException: source already consumed or closed
我有以下程序:
- 以流的形式从文件中读取数据。
- 在上述流上使用 stream.map() 创建一个新流并将其存储在 retVal 中。
- Return retVal.
- 使用 forEach 打印 Stream retval 的值,但我得到异常:IllegalStateException: source already consumed or closed
- 对于函数
workingPopulateFromFileAsStream
,我已经明确地将它再次转换为流,然后它就可以正常工作了。
student.txt 的内容:
Name101,Last101
Name102,Last102
Name103,Last103
Name104,Last104
Name105,Last105
Name106,Last106
Name107,Last107
Name108,Last108
Name109,Last109
Name110,Last110
ClientMainApp.java:
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;
class Student {
String first_name;
String last_name;
public Student(String first_name, String last_name) {
this.first_name = first_name;
this.last_name = last_name;
}
public String toString() {
return this.first_name + " " + this.last_name;
}
}
public class ClientMainApp {
public static void main(String... args) {
String fileName = "student.txt";
ClientMainApp obj = new ClientMainApp();
//Working Code.
Stream<Student> students_stream = obj.workingPopulateFromFileAsStream(fileName);
System.out.println("working: Got Student Stream as " + students_stream);
students_stream.forEach(System.out::println);
System.out.println("==================================================");
// Not Working Code.
Stream<Student> students_stream2 = obj.populateFromFileAsStream2(fileName);
System.out.println("notWorking: Got Student Stream2 as " + students_stream2);
students_stream2.forEach(System.out::println);
}
private Stream<Student> workingPopulateFromFileAsStream(String fileName) {
try (Stream<String> stream = Files.lines(Paths.get(fileName))) {
System.out.println("Working: Got File Stream as " + stream);
Stream<Student> retval =
stream
.map(
st -> {
ArrayList<String> record = new ArrayList<>(Arrays.asList(st.split(",")));
return new Student(record.get(0), record.get(1));
})
.collect(Collectors.toList()).stream(); // This is the code making the difference.
System.out.println("Working: Got Retval Stream as " + retval);
return retval;
} catch (IOException e) {
e.printStackTrace();
}
return null;
}
private Stream<Student> populateFromFileAsStream2(String fileName) {
try (Stream<String> stream = Files.lines(Paths.get(fileName))) {
System.out.println("NotWorking: Got File Stream as " + stream);
Stream<Student> retval =
stream.map(
st -> {
ArrayList<String> record = new ArrayList<>(Arrays.asList(st.split(",")));
return new Student(record.get(0), record.get(1));
});
System.out.println("NotWorking: got Retval Stream as " + retval);
return retval;
} catch (IOException e) {
e.printStackTrace();
}
return null;
}
}
我得到的结果是:
Working: Got File Stream as java.util.stream.ReferencePipeline$Head@3e3abc88
Working: Got Retval Stream as java.util.stream.ReferencePipeline$Head@214c265e
working: Got Student Stream as java.util.stream.ReferencePipeline$Head@214c265e
Name101 Last101
Name102 Last102
Name103 Last103
Name104 Last104
Name105 Last105
Name106 Last106
Name107 Last107
Name108 Last108
Name109 Last109
Name110 Last110
==================================================
NotWorking: Got File Stream as java.util.stream.ReferencePipeline$Head@7699a589
NotWorking: got Retval Stream as java.util.stream.ReferencePipeline@4dd8dc3
notWorking: Got Student Stream2 as java.util.stream.ReferencePipeline@4dd8dc3
Exception in thread "main" java.lang.IllegalStateException: source already consumed or closed
at java.util.stream.AbstractPipeline.sourceSpliterator(AbstractPipeline.java:406)
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:418)
at ClientMainApp.main(ClientMainApp.java:38)
在响应中,我们可以看到 retVal 的流对象与返回调用函数的流对象相同,但我不确定为什么它说它已被消耗。
请注意,如果我在 populateFromFileAsStream2
中使用 forEach 循环,那么它工作正常。只有当函数返回值时,才会报错。
您的非工作版本基本上如下所示:
public Stream<String> foo(Path file) throws IOException {
try (Stream<String> stream = Files.lines(file)) {
return stream.map(...);
}
}
您正在使用 try-with-resources statement。这意味着一旦 try
块退出,Stream
就会关闭,这必须发生才能使方法 return。因此,您在从方法中 returning 之前关闭流。也就是说,我认为您的错误根源在于:
- Create a new stream using stream.map() over the above stream and store it in retVal.
虽然 map
确实会 return 一个新的 Stream
对象,但新对象仍然是同一概念 stream/pipeline 的一部分。换句话说,关闭 stream
引用与关闭 retVal
引用相同。您的工作代码通过 return 一个全新的管道解决了这个问题,但代价是将整个文件缓冲到内存中。
我有以下程序:
- 以流的形式从文件中读取数据。
- 在上述流上使用 stream.map() 创建一个新流并将其存储在 retVal 中。
- Return retVal.
- 使用 forEach 打印 Stream retval 的值,但我得到异常:IllegalStateException: source already consumed or closed
- 对于函数
workingPopulateFromFileAsStream
,我已经明确地将它再次转换为流,然后它就可以正常工作了。
student.txt 的内容:
Name101,Last101
Name102,Last102
Name103,Last103
Name104,Last104
Name105,Last105
Name106,Last106
Name107,Last107
Name108,Last108
Name109,Last109
Name110,Last110
ClientMainApp.java:
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;
class Student {
String first_name;
String last_name;
public Student(String first_name, String last_name) {
this.first_name = first_name;
this.last_name = last_name;
}
public String toString() {
return this.first_name + " " + this.last_name;
}
}
public class ClientMainApp {
public static void main(String... args) {
String fileName = "student.txt";
ClientMainApp obj = new ClientMainApp();
//Working Code.
Stream<Student> students_stream = obj.workingPopulateFromFileAsStream(fileName);
System.out.println("working: Got Student Stream as " + students_stream);
students_stream.forEach(System.out::println);
System.out.println("==================================================");
// Not Working Code.
Stream<Student> students_stream2 = obj.populateFromFileAsStream2(fileName);
System.out.println("notWorking: Got Student Stream2 as " + students_stream2);
students_stream2.forEach(System.out::println);
}
private Stream<Student> workingPopulateFromFileAsStream(String fileName) {
try (Stream<String> stream = Files.lines(Paths.get(fileName))) {
System.out.println("Working: Got File Stream as " + stream);
Stream<Student> retval =
stream
.map(
st -> {
ArrayList<String> record = new ArrayList<>(Arrays.asList(st.split(",")));
return new Student(record.get(0), record.get(1));
})
.collect(Collectors.toList()).stream(); // This is the code making the difference.
System.out.println("Working: Got Retval Stream as " + retval);
return retval;
} catch (IOException e) {
e.printStackTrace();
}
return null;
}
private Stream<Student> populateFromFileAsStream2(String fileName) {
try (Stream<String> stream = Files.lines(Paths.get(fileName))) {
System.out.println("NotWorking: Got File Stream as " + stream);
Stream<Student> retval =
stream.map(
st -> {
ArrayList<String> record = new ArrayList<>(Arrays.asList(st.split(",")));
return new Student(record.get(0), record.get(1));
});
System.out.println("NotWorking: got Retval Stream as " + retval);
return retval;
} catch (IOException e) {
e.printStackTrace();
}
return null;
}
}
我得到的结果是:
Working: Got File Stream as java.util.stream.ReferencePipeline$Head@3e3abc88
Working: Got Retval Stream as java.util.stream.ReferencePipeline$Head@214c265e
working: Got Student Stream as java.util.stream.ReferencePipeline$Head@214c265e
Name101 Last101
Name102 Last102
Name103 Last103
Name104 Last104
Name105 Last105
Name106 Last106
Name107 Last107
Name108 Last108
Name109 Last109
Name110 Last110
==================================================
NotWorking: Got File Stream as java.util.stream.ReferencePipeline$Head@7699a589
NotWorking: got Retval Stream as java.util.stream.ReferencePipeline@4dd8dc3
notWorking: Got Student Stream2 as java.util.stream.ReferencePipeline@4dd8dc3
Exception in thread "main" java.lang.IllegalStateException: source already consumed or closed
at java.util.stream.AbstractPipeline.sourceSpliterator(AbstractPipeline.java:406)
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:418)
at ClientMainApp.main(ClientMainApp.java:38)
在响应中,我们可以看到 retVal 的流对象与返回调用函数的流对象相同,但我不确定为什么它说它已被消耗。
请注意,如果我在 populateFromFileAsStream2
中使用 forEach 循环,那么它工作正常。只有当函数返回值时,才会报错。
您的非工作版本基本上如下所示:
public Stream<String> foo(Path file) throws IOException {
try (Stream<String> stream = Files.lines(file)) {
return stream.map(...);
}
}
您正在使用 try-with-resources statement。这意味着一旦 try
块退出,Stream
就会关闭,这必须发生才能使方法 return。因此,您在从方法中 returning 之前关闭流。也就是说,我认为您的错误根源在于:
- Create a new stream using stream.map() over the above stream and store it in retVal.
虽然 map
确实会 return 一个新的 Stream
对象,但新对象仍然是同一概念 stream/pipeline 的一部分。换句话说,关闭 stream
引用与关闭 retVal
引用相同。您的工作代码通过 return 一个全新的管道解决了这个问题,但代价是将整个文件缓冲到内存中。