Java8 资源管理与 RxJava 扩展详解

通过前面的章节，我们已经学习了如何使用 RxJava 的可观察性。我们一直在使用许多不同的运算符和factory方法。factory方法是各种Observable实例的来源，具有不同的行为和排放源。另一方面，通过使用操作符，我们已经围绕这些观测值构建了复杂的逻辑。

在本章中，我们将学习如何创建我们自己的factory方法，这些方法将能够管理它们的源资源。为了做到这一点，我们需要一种管理和处置资源的方法。我们对源文件、HTTP 请求、文件夹或内存中的数据创建并使用了多种方法。但是他们中的一些人不清理他们的资源。例如，HTTP 请求 observable 需要一个CloseableHttpAsyncClient实例；我们创建了一个接收它的方法，并将它的管理留给用户。现在是学习如何自动管理和清理源数据的时候了，这些数据封装在我们的factory方法中。

我们还将学习如何编写自己的运算符。Java 不是一种动态语言，这就是为什么我们不会添加操作符作为Observable类的方法。有一种方法可以将它们插入到可观察的行动链中，我们将在本章中看到这一点。

本章涵盖的主题包括：

如果我们回顾一下我们在第 6 章、使用与调度器和第 5 章、组合符、条件和错误处理中使用的 HTTP 请求方法，它有以下签名：Observable<Map> requestJson(HttpAsyncClient client, String url)。

我们不只是调用一个向 URL 发出请求并以 JSON 形式返回响应的方法，而是创建一个HttpAsyncClient实例，必须启动它并将其传递给requestJson()方法。但还有更多：我们需要在读取结果后关闭客户端，并且由于可观察对象是异步，我们需要等待其OnCompleted通知，然后进行关闭。这非常复杂，应该改变。读取文件的Observable需要创建流/读卡器/频道，并在所有订户取消订阅时关闭它们。从数据库发送数据的Observable应该设置，然后关闭读取完成后使用的所有连接、语句和结果集。对于HttpAsyncClient对象也是如此。它是我们用来打开到远程服务器的连接的资源；我们的 observable 应该在所有内容都被读取并且所有订阅者都不再订阅之后将其清理干净。

让我们回答这个问题：requestJson()方法为什么需要这个HttpAsyncClient对象？答案是我们对 HTTP 请求使用了一个 RxJava 模块。其代码如下：

ObservableHttp
  .createGet(url, client)
  .toObservable();

这段代码创建请求，而代码需要客户端，因此我们需要客户端来创建我们的Observable实例。我们不能更改此代码，因为更改它意味着自己编写 HTTP 请求，这并不好。已经有一家图书馆为我们做这件事了。我们必须在订阅时使用提供HttpAsyncClient实例的东西，并在取消订阅时从中处理。有一种方法可以做到这一点：using()工厂方法。

介绍可观察的使用方法

Observable.using方法的签名如下：

public final static <T, Resource> Observable<T> using(
  final Func0<Resource> resourceFactory,
  final Func1<? super Resource, ? extends Observable<? extends T>> observableFactory,
  final Action1<? super Resource> disposeAction
)

这看起来很复杂，但再看一眼就不难理解了。让我们来看看下面的描述：

  • 它的第一个参数是Func0<Resource> resourceFactory，一个创建Resource对象的函数（这里Resource是任意对象；它不是接口或类，而是类型参数的名称）。我们的工作是实现资源创建。
  • 第二个参数Func1<? super Resource, ? extends Observable<? extends T>> observableFactory是一个函数，它接收Resource对象并返回Observable实例。此函数将使用我们已经通过第一个参数创建的Resource对象调用。我们可以使用此资源创建我们的Observable实例。
  • 处理Resource对象时调用Action1<? super Resource> disposeAction参数。它接收由resourceFactory参数创建的Resource对象（用于创建Observable实例），我们的工作就是处理它。这是在取消订阅时调用的。

我们现在可以创建一个函数，发出 HTTP 请求，而无需将HttpAsyncClient对象传递给它。我们有公用设施，可以根据需要创建和处理它。让我们实现以下功能：

// (1)
public Observable<ObservableHttpResponse> request(String url) {
  Func0<CloseableHttpAsyncClient> resourceFactory = () -> {
    CloseableHttpAsyncClient client = HttpAsyncClients.createDefault(); // (2)
 client.start();
    System.out.println(
      Thread.currentThread().getName() +
      " : Created and started the client."
    );
    return client;
  };
  Func1<HttpAsyncClient, Observable<ObservableHttpResponse>> observableFactory = (client) -> { // (3)
    System.out.println(
      Thread.currentThread().getName() + " : About to create Observable."
    );
    return ObservableHttp.createGet(url, client).toObservable();
  };
  Action1<CloseableHttpAsyncClient> disposeAction = (client) -> {
    try { // (4)
      System.out.println(
        Thread.currentThread().getName() + " : Closing the client."
      );
      client.close();
    }
    catch (IOException e) {}
  };
  return Observable.using( // (5)
 resourceFactory,
 observableFactory,
 disposeAction
 );
}

这个方法并不难理解。让我们来分解一下：

  1. 该方法签名简单；它只有一个参数URL。方法的调用方不需要创建和管理CloseableHttpAsyncClient实例的生命周期。返回一个能够发出ObservableHttpResponse响应并完成的Observable实例。getJson()方法可以使用它将ObservableHttpResponse响应转换为表示 JSON 的Map 实例，同样不需要传递客户端。
  2. resourceFactorylambda 很简单；它创建一个默认的CloseableHttpAsyncClient实例并启动它。调用时，它将返回一个初始化的 HTTP客户端，该客户端能够请求远程服务器数据。我们输出客户端已准备好进行调试。
  3. observableFactory函数可以访问resourceFactory函数创建的CloseableHttpAsyncClient实例，因此它使用它和传递的URL来构造生成的Observable实例。此是通过 RxJava 的rxjava-apache-http模块 API（完成的 https://github.com/ReactiveX/RxApacheHttp 。我们输出我们正在做的事情。
  4. disposeAction函数接收用于创建Observable实例的CloseableHttpAsyncClient对象，并关闭实例。同样地，我们将一条消息打印到标准输出，表示我们将要这样做。
  5. 借助using()工厂方法，我们返回 HTTP请求Observable实例。这不会触发三个 lambda 中的任何一个。订阅返回的Observable实例将调用resourceFactory函数，然后调用observableFactory函数。

这就是我们如何实现一个能够管理自身资源的Observable实例。让我们看看它是如何使用的：

String url = "https://api.github.com/orgs/ReactiveX/repos";

Observable<ObservableHttpResponse> response = request(url);

System.out.println("Not yet subscribed.");

Observable<String> stringResponse = response
.<String>flatMap(resp -> resp.getContent()
.map(bytes -> new String(bytes, java.nio.charset.StandardCharsets.UTF_8))
.retry(5)

.map(String::trim);

System.out.println("Subscribe 1:");
System.out.println(stringResponse.toBlocking().first());

System.out.println("Subscribe 2:");
System.out.println(stringResponse.toBlocking().first());

我们使用新的request()方法列出*ReactiveX*组织机构的存储库。我们只需将 URL 传递给它，就会得到Observable响应。在我们订阅之前，不会分配任何资源，也不会执行任何请求，因此我们打印您尚未订阅。

stringResponse可观察对象包含逻辑并将原始ObservableHttpResponse对象转换为String。但是，没有分配任何资源，也没有发送任何请求。

我们使用BlockingObservable类的first()方法订阅Observable实例并等待其结果。我们将响应检索为String并将其输出。现在，分配资源并发出请求。获取数据后，BlockingObservable实例封装的subscriber会自动取消订阅，因此使用的资源（HTTP 客户端）会被释放。我们再订阅一次，看看接下来会发生什么。

让我们看看这个程序的输出：

Not yet subscribed.
Subscribe 1:
main : Created and started the client.
main : About to create Observable.
[{"id":7268616,"name":"Rx.rb","full_name":"ReactiveX/Rx.rb",...
Subscribe 2:
I/O dispatcher 1 : Closing the client.
main : Created and started the client.
main : About to create Observable.
I/O dispatcher 5 : Closing the client.
[{"id":7268616,"name":"Rx.rb","full_name":"ReactiveX/Rx.rb",...

因此，当我们订阅网站时，HTTP 客户端和Observable实例将使用我们的工厂 lambdas 创建。创建在当前主线程上执行。请求已提出并打印（此处裁剪）。客户端在 IO 线程上被释放，请求在Observable实例完成执行时执行。

当第二次订阅时，我们从一开始就经历相同的过程；我们分配资源，创建Observable实例并处理资源。这是因为using()方法的工作方式是为每个订阅分配一个资源。我们可以使用不同的技术在下一个订阅上重用相同的结果，而不是发出新的请求并为其分配资源。例如，我们可以对多个订户或一个Subject实例重用CompositeSubscription方法。但是，有一种更简单的方法可以重用获取的下一个订阅响应。

我们可以使用缓存将响应缓存在内存中，然后在下一次订阅时使用缓存数据，而不是再次请求远程服务器。

让我们将代码更改为如下所示：

String url = "https://api.github.com/orgs/ReactiveX/repos";
Observable<ObservableHttpResponse> response = request(url);

System.out.println("Not yet subscribed.");
Observable<String> stringResponse = response
.flatMap(resp -> resp.getContent()
.map(bytes -> new String(bytes)))
.retry(5)
.cast(String.class)
.map(String::trim)
.cache();

System.out.println("Subscribe 1:");
System.out.println(stringResponse.toBlocking().first());

System.out.println("Subscribe 2:");
System.out.println(stringResponse.toBlocking().first());

在stringResponse链末端调用的cache()操作符将缓存subscribers之后所有的响应，该响应由string表示。因此，这次的输出将是：

Not yet subscribed.
Subscribe 1:
main : Created and started the client.
main : About to create Observable.
[{"id":7268616,"name":"Rx.rb",...
I/O dispatcher 1 : Closing the client.
Subscribe 2:
[{"id":7268616,"name":"Rx.rb",...

现在，我们可以通过我们的程序重用我们的stringResponse``Observable实例，而无需进行额外的资源分配和请求。

注

演示源代码可在找到 https://github.com/meddle0x53/learning-rxjava/blob/master/src/main/java/com/packtpub/reactive/chapter08/ResourceManagement.java 。

最后，requestJson()方法可以这样实现：

public Observable<Map> requestJson(String url) {
Observable<String> rawResponse = request(url)

....

return Observable.amb(fromCache(url), response);
}

更简单，并且资源自动管理（资源，http 客户端自动创建和销毁），该方法也实现了自己的缓存功能（我们在第 5 章、组合符、条件和错误处理中实现了它）。

注

通过本书开发的所有创建Observable实例的方法都可以在[找到 https://github.com/meddle0x53/learning-rxjava/blob/master/src/main/java/com/packtpub/reactive/common/CreateObservable.java 类](https://github.com/meddle0x53/learning-rxjava/blob/master/src/main/java/com/packtpub/reactive/common/CreateObservable.java class)包含在源代码中。您也可以在那里找到requestJson()方法的文件缓存实现。

有了这个，我们能够扩展 RxJava，创建我们自己的工厂方法，使Observable实例依赖于任意数据源。

本章的下一节将展示如何将我们自己的逻辑放入Observable操作符链中。

在学习了并使用了这么多各种运算符之后，我们准备编写自己的运算符。Observable类有一个名为lift的运算符。它接收Operator接口的一个实例。这个接口只是一个扩展了Func1<Subscriber<? super R>, Subscriber<? super T>>接口的空接口。这意味着我们甚至可以将 lambda 作为操作符传递。

学习如何使用lift运算符的最佳方法是编写一个示例。让我们创建一个操作符，为每个发出的项添加一个顺序索引（当然，这在没有专用操作符的情况下是可行的）。这样，我们将能够生成索引项。为此，我们需要一个存储项及其索引的类。让我们创建一个更通用的类，名为Pair：

public class Pair<L, R> {
  final L left;
  final R right;

public Pair(L left, R right) {
    this.left = left;
    this.right = right;
  }

  public L getLeft() {
    return left;
  }

public R getRight() {
    return right;
  }

  @Override
  public String toString() {
    return String.format("%s : %s", this.left, this.right);
  }

// hashCode and equals omitted

}'

这个类的实例是非常简单的不可变对象，包含两个任意对象。在我们的例子中，左字段将是Long类型的索引，右字段将是发出的项目。与任何不可变类一样，Pair类包含hashCode()和equals()方法的实现。

以下是操作员的代码：

public class Indexed<T> implements Operator<Pair<Long, T>, T> {
  private final long initialIndex;
  public Indexed() {
    this(0L);
  }
  public Indexed(long initial) {
    this. initialIndex = initial;
  }
  @Override
  public Subscriber<? super T> call(Subscriber<? super Pair<Long, T>> s) {
 return new Subscriber<T>(s) {
      private long index = initialIndex;
 @Override
 public void onCompleted() {
 s.onCompleted();
 }
 @Override
 public void onError(Throwable e) {
 s.onError(e);
 }
 @Override
 public void onNext(T t) {
 s.onNext(new Pair<Long, T>(index++, t));
 }
 };
 }
}

Operator接口的call()方法有一个参数Subscriber实例。此实例将订阅操作符返回的可观察对象。方法返回一个新的Subscriber实例，该实例将订阅调用lift()操作符的可观察对象。我们可以更改其中所有通知的数据，这就是我们编写自己的操作员逻辑的方式。

Indexed类有一个状态-index。默认初始值为0，但有构造函数可以创建任意初始值的Indexed实例。我们的运营商将OnError和OnCompleted通知委托给订户，但不做任何更改。有趣的方法是onNext()。它通过创建传入项的Pair实例和index字段的当前值来修改传入项。之后，index增加。这样，下一项将使用递增的index并再次递增。

现在，我们有了第一个接线员。让我们编写一个单元测试来展示其行为：

@Test
public void testGeneratesSequentialIndexes() {
  Observable<Pair<Long, String>> observable = Observable
    .just("a", "b", "c", "d", "e")
    .lift(new Indexed<String>());
  List<Pair<Long, String>> expected = Arrays.asList(
    new Pair<Long, String>(0L, "a"),
    new Pair<Long, String>(1L, "b"),
    new Pair<Long, String>(2L, "c"),
    new Pair<Long, String>(3L, "d"),
    new Pair<Long, String>(4L, "e")
  );
  List<Pair<Long, String>> actual = observable
    .toList()
    .toBlocking().
    single();
  assertEquals(expected, actual);
  // Assert that it is the same result for a second subscribtion.
  TestSubscriber<Pair<Long, String>> testSubscriber = new TestSubscriber<Pair<Long, String>>();
  observable.subscribe(testSubscriber);
  testSubscriber.assertReceivedOnNext(expected);
}

测试发出从'a'到'e'的字母，使用lift()操作符将我们的Indexed操作符实现插入到可观察链中。我们希望列表中有五个从零开始的序列号索引和字母的Pair实例。我们使用toList().toBlocking().single()技术检索实际排放项目列表，并断言它们等于预期排放量。因为Pair实例定义了hashCode()和equals()方法，我们可以比较Pair实例，所以测试通过。如果我们第二次订阅，则Indexed操作员应提供来自初始索引0的索引。使用一个TestSubscriber实例，我们这样做，并断言这些字母是索引的，从0开始。

注

Indexed操作员的代码位于https://github.com/meddle0x53/learning-rxjava/blob/master/src/main/java/com/packtpub/reactive/chapter08/Lift.java 以及在测试其行为的单元测试 https://github.com/meddle0x53/learning-rxjava/blob/master/src/test/java/com/packtpub/reactive/chapter08/IndexedTest.java 。

使用lift()操作符和不同的Operator实现，我们可以编写自己的操作符，对发出序列的每一项进行操作。但在大多数情况下，我们将能够在不创建新操作符的情况下实现我们的逻辑。例如，索引行为可以通过多种方式实现，其中一种是通过压缩和Observable.range方法，如下所示：

Observable<Pair<Long, String>> indexed = Observable.zip(
  Observable.just("a", "b", "c", "d", "e"),
  Observable.range(0, 100),
  (s, i) -> new Pair<Long, String>((long) i, s)
);
subscribePrint(indexed, "Indexed, no lift");

实现一个新操作符有许多陷阱，比如链接订阅、支持背压和重用变量。如果可能，我们应该尝试组合现有的操作符，这些操作符是由经验丰富的 RxJava 贡献者编写的。因此，在某些情况下，一个操作符转换Observable本身是一个更好的主意，例如，将多个操作符作为一个操作符应用于它。为此，我们可以使用合成操作符compose()。

compose()运算符有一个类型为Transformer的参数。与和Operator一样，Transformer接口是一个扩展Func1的空接口（这种方法隐藏了使用Func1所涉及的类型的复杂性）。区别在于它扩展了Func1<Observable<T>, Observable<R>>方法，因此它转换了Observable而不是Subscriber。这意味着，它不是对源可观察到的每个单独项目进行操作，而是直接对源进行操作。

我们可以通过一个例子来说明这个操作符和Transformer接口的使用。首先，我们将创建一个Transformer实现：

public class OddFilter<T> implements Transformer<T, T> {
  @Override
  public Observable<T> call(Observable<T> observable) {
    return observable
      .lift(new Indexed<T>(1L))
      .filter(pair -> pair.getLeft() % 2 == 1)
      .map(pair -> pair.getRight());
  }
}

该实现的思想是根据可观测信号的传入顺序过滤其发射。它对整个序列进行操作，使用我们的Indexed操作符为每个项目添加索引。之后，它过滤索引为奇数的Pair实例，并从过滤后的Pair实例中检索原始项。这样，只有位于奇数位置的发射序列的成员才能到达订阅者。

再次让我们编写一个单元测试，确保新的OddFilter变压器以正确的方式运行：

@Test
public void testFiltersOddOfTheSequence() {
  Observable<String> tested = Observable
    .just("One", "Two", "Three", "Four", "Five", "June", "July")
    .compose(new OddFilter<String>());
  List<String> expected =
    Arrays.asList("One", "Three", "Five", "July");
  List<String> actual = tested
    .toList()
    .toBlocking()
    .single();
  assertEquals(expected, actual);
}

正如您所看到的，我们的OddFilter类的一个实例被传递给compose()操作符，方式被应用于range()工厂方法创建的可观察对象。可见光发射七根弦。如果OddFilter实现工作正常，它应该过滤掉奇数位置发出的字符串。

注

OddFilter类的源代码可在找到 https://github.com/meddle0x53/learning-rxjava/blob/master/src/main/java/com/packtpub/reactive/chapter08/Compose.java 。单元测试测试可在查看/下载 https://github.com/meddle0x53/learning-rxjava/blob/master/src/test/java/com/packtpub/reactive/chapter08/IndexedTest.java 。

关于实施自定义运算符的更多信息，请参见：https://github.com/ReactiveX/RxJava/wiki/Implementing-Your-Own-Operators 。如果您在 Groovy 这样的动态语言中使用 RxJava，您将能够用新方法扩展Observable类，或者您可以将 RxJava 与 Xtend 结合使用，Xtend 是 Java 灵活的方言。参见http://mnmlst-dvlpr.blogspot.de/2014/07/rxjava-and-xtend.html 。

创建我们自己的运算符和依赖于资源的Observable实例为我们在Observable类周围创建逻辑提供了无限的可能性。我们能够将每个数据源转换为一个Observable实例，并以多种不同的方式转换传入的数据。

我希望这本书涵盖 RxJava 最有趣和最重要的部分。如果我遗漏了一些重要的内容，请参阅中的文档 https://github.com/ReactiveX/RxJava/wiki 是网络上最好的产品之一。。进一步阅读请特别参考本节：https://github.com/ReactiveX/RxJava/wiki/Additional-Reading 。

我试图构建代码和想法，并在章节中以小迭代的方式提供它们。第一章和第二章更具意识形态性；他们向读者介绍了函数式编程和反应式编程的基本思想，第二章试图建立Observable类的起源。第三章为读者提供了创建各种不同Observable实例的方法。第四章和第五章教我们如何围绕这些Observable实例编写逻辑，第六章为该逻辑添加了多线程。第七章是单元测试读者已经学会编写的逻辑，第八章试图进一步扩展此逻辑的功能。

我希望你，读者，觉得这本书有用。别忘了，RxJava 只是一个工具。重要的是你的知识和思想。

教程来源于Github，感谢apachecn大佬的无私奉献，致敬！

技术教程推荐

React实战进阶45讲 -〔王沛〕

从0开始做增长 -〔刘津〕

深入拆解Tomcat & Jetty -〔李号双〕

DevOps实战笔记 -〔石雪峰〕

系统性能调优必知必会 -〔陶辉〕

体验设计案例课 -〔炒炒〕

攻克视频技术 -〔李江〕

大厂设计进阶实战课 -〔小乔〕

深入浅出可观测性 -〔翁一磊〕

休闲君评测 |  TOOLFK工具网 |  古诗文网 |  良许Linux教程网 |  Python之禅 |  红色石头 |  入党申请书 | 

© 2020 Hi LearnFk 意见反馈 免责声明 关于我们 桂ICP备11002319号-3 友链&广告位+QQ: 1963612630