我是RxJava2的新手，不确定我是否正确使用它。
我有几个输入流正在从TCP流中读取数据，并且正在将数据读取到byte[]中。然后，我拿起byte[]，为每个InputStream创建一个Flowable，并将单个Flowable合并为一个。
我希望能够关闭 1InputStream但仍有合并的Flowable继续从未关闭的InputStream读取。
目前，我让它正确读取 2 个输入流，但是当我调用close时，它会导致异常，java.net.SocketException: Socket closed被抛出。
我的问题是，如何正确关闭输入流或可流动，以便我可以同时继续从另一个输入流读取。
InputStream inputStream = response.body().asInputStream();
InputStream inputStream2 = response2.body().asInputStream();
Flowable<byte[]> flowable = Bytes.from(inputStream)
.distinct();
Flowable<byte[]> flowable2 = Bytes.from(inputStream2)
.distinct();
Flowable.merge(flowable.subscribeOn(Schedulers.newThread()),
flowable2.subscribeOn(Schedulers.newThread()))
.subscribe(s -> System.out.println("Data: " + new String(s)),
e -> System.out.println("Error: " + e.getLocalizedMessage() + " : " + e.getMessage()),
() -> System.out.println("Complete!"));
//Calling this results in an exception being thrown.
inputSteam2.close();
堆栈跟踪：
io.reactivex.exceptions.UndeliverableException: java.net.SocketException: Socket closed
at io.reactivex.plugins.RxJavaPlugins.onError(RxJavaPlugins.java:349)
at io.reactivex.internal.operators.flowable.FlowableFlatMap$MergeSubscriber.innerError(FlowableFlatMap.java:604)
at io.reactivex.internal.operators.flowable.FlowableFlatMap$InnerSubscriber.onError(FlowableFlatMap.java:665)
at io.reactivex.internal.operators.flowable.FlowableSubscribeOn$SubscribeOnSubscriber.onError(FlowableSubscribeOn.java:102)
at io.reactivex.internal.operators.flowable.FlowableGenerate$GeneratorSubscription.onError(FlowableGenerate.java:189)
at io.reactivex.internal.operators.flowable.FlowableGenerate$GeneratorSubscription.request(FlowableGenerate.java:114)
at io.reactivex.internal.operators.flowable.FlowableSubscribeOn$SubscribeOnSubscriber.requestUpstream(FlowableSubscribeOn.java:133)
at io.reactivex.internal.operators.flowable.FlowableSubscribeOn$SubscribeOnSubscriber.onSubscribe(FlowableSubscribeOn.java:90)
at io.reactivex.internal.operators.flowable.FlowableGenerate.subscribeActual(FlowableGenerate.java:52)
at io.reactivex.Flowable.subscribe(Flowable.java:12986)
at io.reactivex.Flowable.subscribe(Flowable.java:12932)
at io.reactivex.internal.operators.flowable.FlowableSubscribeOn$SubscribeOnSubscriber.run(FlowableSubscribeOn.java:82)
at io.reactivex.internal.schedulers.ScheduledRunnable.run(ScheduledRunnable.java:61)
at io.reactivex.internal.schedulers.ScheduledRunnable.call(ScheduledRunnable.java:52)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:748)
Caused by: java.net.SocketException: Socket closed
at java.net.SocketInputStream.socketRead0(Native Method)
at java.net.SocketInputStream.socketRead(SocketInputStream.java:116)
at java.net.SocketInputStream.read(SocketInputStream.java:171)
at java.net.SocketInputStream.read(SocketInputStream.java:141)
at sun.security.ssl.InputRecord.readFully(InputRecord.java:465)
at sun.security.ssl.InputRecord.read(InputRecord.java:503)
at sun.security.ssl.SSLSocketImpl.readRecord(SSLSocketImpl.java:973)
at sun.security.ssl.SSLSocketImpl.readDataRecord(SSLSocketImpl.java:930)
at sun.security.ssl.AppInputStream.read(AppInputStream.java:105)
at okio.Okio$2.read(Okio.java:140)
at okio.AsyncTimeout$2.read(AsyncTimeout.java:238)
at okio.RealBufferedSource.read(RealBufferedSource.java:45)
at okhttp3.internal.http.Http1xStream$UnknownLengthSource.read(Http1xStream.java:476)
at okio.RealBufferedSource$1.read(RealBufferedSource.java:386)
at java.io.InputStream.read(InputStream.java:101)
at com.github.davidmoten.rx2.Bytes$1.accept(Bytes.java:47)
at com.github.davidmoten.rx2.Bytes$1.accept(Bytes.java:43)
at io.reactivex.internal.operators.flowable.FlowableInternalHelper$SimpleGenerator.apply(FlowableInternalHelper.java:44)
at io.reactivex.internal.operators.flowable.FlowableInternalHelper$SimpleGenerator.apply(FlowableInternalHelper.java:35)
at io.reactivex.internal.operators.flowable.FlowableGenerate$GeneratorSubscription.request(FlowableGenerate.java:109)
... 14 more
Exception in thread "RxNewThreadScheduler-3" io.reactivex.exceptions.UndeliverableException: java.net.SocketException: Socket closed
at io.reactivex.plugins.RxJavaPlugins.onError(RxJavaPlugins.java:349)
at io.reactivex.internal.operators.flowable.FlowableFlatMap$MergeSubscriber.innerError(FlowableFlatMap.java:604)
at io.reactivex.internal.operators.flowable.FlowableFlatMap$InnerSubscriber.onError(FlowableFlatMap.java:665)
at io.reactivex.internal.operators.flowable.FlowableSubscribeOn$SubscribeOnSubscriber.onError(FlowableSubscribeOn.java:102)
at io.reactivex.internal.operators.flowable.FlowableGenerate$GeneratorSubscription.onError(FlowableGenerate.java:189)
at io.reactivex.internal.operators.flowable.FlowableGenerate$GeneratorSubscription.request(FlowableGenerate.java:114)
at io.reactivex.internal.operators.flowable.FlowableSubscribeOn$SubscribeOnSubscriber.requestUpstream(FlowableSubscribeOn.java:133)
at io.reactivex.internal.operators.flowable.FlowableSubscribeOn$SubscribeOnSubscriber.onSubscribe(FlowableSubscribeOn.java:90)
at io.reactivex.internal.operators.flowable.FlowableGenerate.subscribeActual(FlowableGenerate.java:52)
at io.reactivex.Flowable.subscribe(Flowable.java:12986)
at io.reactivex.Flowable.subscribe(Flowable.java:12932)
at io.reactivex.internal.operators.flowable.FlowableSubscribeOn$SubscribeOnSubscriber.run(FlowableSubscribeOn.java:82)
at io.reactivex.internal.schedulers.ScheduledRunnable.run(ScheduledRunnable.java:61)
at io.reactivex.internal.schedulers.ScheduledRunnable.call(ScheduledRunnable.java:52)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:748)
我能够通过创建自己的运算符并从中创建Flowable来成功关闭流。
这个类是一个新的 RxJava 运算符，它实现并记录我关闭InputStream时得到的异常，这对于我的用例来说没问题。
public class MyFlowableOperator implements FlowableOperator {
@Override
public Subscriber apply(Subscriber subscriber) throws Exception {
return new Subscriber() {
@Override
public void onSubscribe(Subscription s) {
subscriber.onSubscribe(s);
}
@Override
public void onNext(Object o) {
subscriber.onNext(o);
}
@Override
public void onError(Throwable t) {
if(t instanceof SocketException) {
log.debug("Input Stream Closed");
} else {
subscriber.onError(t);
}
}
@Override
public void onComplete() {
subscriber.onComplete();
}
};
}
}
以下是我如何声明 RxJava 运算符的新实例，并在从InputStream创建Flowable时使用它
// Private method to utilize the new RxJava Operator
private Subscriber getFlows(Subscriber subscriber) {
try {
return new MyFlowableOperator().apply(subscriber);
} catch (Exception e) {
log.error(e.getMessage());
}
return null;
}
创建Flowable本身时，我使用lift方法来调用我的方法并返回一个带有错误处理的Flowable。
// Declaration of the Flowables
Flowable<byte[]> flowable = Bytes.from(inputStream)
.subscribeOn(Schedulers.newThread())
.distinct()
.lift(this::getFlows);
Flowable<byte[]> flowable2 = Bytes.from(inputStream2)
.subscribeOn(Schedulers.newThread())
.distinct()
.lift(this::getFlows);