如果关闭，RxJava 2.x 如何继续合并 InputStreams 的 Flowable

Question

我是 RxJava2 的新手，不确定我是否正确使用它。

我有几个从 TCP 流读取数据的 InputStreams，并且正在将数据读入 byte[]。然后，我将 byte[] 并为每个 InputStream 创建一个 Flowable，并将单个 Flowable 合并为一个。

我希望能够关闭 1 InputStream，但仍然有合并的 Flowable 继续从未关闭的 InputStream 读取。

目前，我可以正确地从 2 个 InputStreams 中读取数据，但是当我调用 close 时，它会导致异常，java.net.SocketException: Socket closed 被抛出。

我的问题是，如何正确关闭 InputStream 或 Flowable，以便我可以同时继续从其他 InputStream 读取。

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.access1(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.read(Okio.java:140)
  at okio.AsyncTimeout.read(AsyncTimeout.java:238)
  at okio.RealBufferedSource.read(RealBufferedSource.java:45)
  at okhttp3.internal.http.Http1xStream$UnknownLengthSource.read(Http1xStream.java:476)
  at okio.RealBufferedSource.read(RealBufferedSource.java:386)
  at java.io.InputStream.read(InputStream.java:101)
  at com.github.davidmoten.rx2.Bytes.accept(Bytes.java:47)
  at com.github.davidmoten.rx2.Bytes.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.access1(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)

Answer 1

我能够通过创建自己的运算符并从中创建 Flowable 来成功关闭流。

这个 class 是一个新的 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 方法来调用我的方法，return 使用 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);

如果关闭，RxJava 2.x 如何继续合并 InputStreams 的 Flowable

How to continue with merged Flowable of InputStreams if one is closed, RxJava 2.x

java

inputstream

java-8

rx-java2