Introduction to RxJava – RxJava简介

最后修改: 2017年 9月 11日

中文/混合/英文(键盘快捷键：t)

1. Overview

1.概述

In this article, we’re going to focus on using Reactive Extensions (Rx) in Java to compose and consume sequences of data.

在这篇文章中，我们将专注于使用Java中的Reactive Extensions（Rx）来编排和消费数据序列。

At a glance, the API may look similar to Java 8 Streams, but in fact, it is much more flexible and fluent, making it a powerful programming paradigm.

乍一看，该API可能与Java 8 Streams相似，但事实上，它更灵活、更流畅，使其成为一个强大的编程范式。

If you want to read more about RxJava, check out this writeup.

如果你想阅读更多关于RxJava的信息，请查看这篇报道。

2. Setup

2.设置

To use RxJava in our Maven project, we’ll need to add the following dependency to our pom.xml:

要在我们的Maven项目中使用RxJava，我们需要在pom.xml中添加以下依赖项：。

<dependency>
    <groupId>io.reactivex</groupId>
    <artifactId>rxjava</artifactId>
    <version>${rx.java.version}</version>
</dependency>

Or, for a Gradle project:

或者，对于一个Gradle项目。

compile 'io.reactivex.rxjava:rxjava:x.y.z'

3. Functional Reactive Concepts

3.功能性的反应性概念

On one side, functional programming is the process of building software by composing pure functions, avoiding shared state, mutable data, and side-effects.

一方面，函数式编程是通过组成纯函数来构建软件的过程，避免了共享状态、可变数据和副作用。

On the other side, reactive programming is an asynchronous programming paradigm concerned with data streams and the propagation of change.

另一方面，反应式编程是一种关注数据流和变化传播的异步编程范式。

Together, functional reactive programming forms a combination of functional and reactive techniques that can represent an elegant approach to event-driven programming – with values that change over time and where the consumer reacts to the data as it comes in.

一起，功能反应式编程形成了功能和反应式技术的组合，可以代表一种优雅的事件驱动编程方法–其值随时间变化，消费者在数据进来时对其做出反应。

This technology brings together different implementations of its core principles, some authors came up with a document that defines the common vocabulary for describing the new type of applications.

这项技术汇集了其核心原则的不同实现，一些作者想出了一份文件，定义了描述新型应用的通用词汇。

3.1. Reactive Manifesto

3.1.反应式宣言

The Reactive Manifesto is an online document that lays out a high standard for applications within the software development industry. Simply put, reactive systems are:

反应式宣言是一份在线文件，为软件开发行业内的应用规定了一个高标准。简单地说，反应式系统是

  • Responsive – systems should respond in a timely manner
  • Message Driven – systems should use async message-passing between components to ensure loose coupling
  • Elastic – systems should stay responsive under high load
  • Resilient – systems should stay responsive when some components fail

4. Observables

4.可观察因素

There are two key types to understand when working with Rx:

在使用Rx时，有两种关键类型需要了解：Rx：。

  • Observable represents any object that can get data from a data source and whose state may be of interest in a way that other objects may register an interest
  • An observer is any object that wishes to be notified when the state of another object changes

An observer subscribes to an Observable sequence. The sequence sends items to the observer one at a time.

一个observer订阅了一个Observable序列。该序列每次向观察者发送一个项目。

The observer handles each one before processing the next one. If many events come in asynchronously, they must be stored in a queue or dropped.

观察者在处理下一个事件之前处理每一个事件。如果许多事件是异步进来的，它们必须被存储在队列中或被丢弃。

In Rx, an observer will never be called with an item out of order or called before the callback has returned for the previous item.

在Rx中，observer将永远不会被不按顺序地调用，或者在前一个项目的回调返回之前被调用。

4.1. Types of Observable

4.1.可观察的类型

There are two types:

有两种类型。

  • Non-Blocking – asynchronous execution is supported and is allowed to unsubscribe at any point in the event stream. On this article, we’ll focus mostly on this kind of type
  • Blocking – all onNext observer calls will be synchronous, and it is not possible to unsubscribe in the middle of an event stream. We can always convert an Observable into a Blocking Observable, using the method toBlocking:
BlockingObservable<String> blockingObservable = observable.toBlocking();

4.2. Operators

4.2.操作员

An operator is a function that takes one Observable (the source) as its first argument and returns another Observable (the destination). Then for every item that the source observable emits, it will apply a function to that item, and then emit the result on the destination Observable.

一个operator是一个函数，它将一个Observable（源）作为它的第一个参数，并返回另一个Observable（目的）。那么对于源观察器所发射的每一个项目，它将对该项目应用一个函数，然后在目的Observable上发射结果。

Operators can be chained together to create complex data flows that filter event based on certain criteria. Multiple operators can be applied to the same observable.

操作符可被连锁起来，以创建复杂的数据流，根据某些标准过滤事件。多个操作符可应用于同一个可观察物。

It is not difficult to get into a situation in which an Observable is emitting items faster than an operator or observer can consume them. You can read more about back-pressure here.

不难发现，Observable发射项目的速度比operator或observer消耗它们的速度快。你可以在这里阅读更多关于背压的内容。

4.3. Create Observable

4.3.创建可观察变量

The basic operator just produces an Observable that emits a single generic instance before completing, the String “Hello”. When we want to get information out of an Observable, we implement an observer interface and then call subscribe on the desired Observable:

基本操作符只是产生了一个Observable，在完成之前发射了一个单一的通用实例，即字符串“Hello”。当我们想从Observable中获取信息时，我们实现一个observer接口，然后在所需的Observable上调用 subscribe：。

Observable<String> observable = Observable.just("Hello");
observable.subscribe(s -> result = s);
 
assertTrue(result.equals("Hello"));

4.4. OnNext, OnError, and OnCompleted

4.4.OnNext, OnError, and OnCompleted

There are three methods on the observer interface that we want to know about:

在observer接口上有三个方法，我们想了解一下。

  1. OnNext is called on our observer each time a new event is published to the attached Observable. This is the method where we’ll perform some action on each event
  2. OnCompleted is called when the sequence of events associated with an Observable is complete, indicating that we should not expect any more onNext calls on our observer
  3. OnError is called when an unhandled exception is thrown during the RxJava framework code or our event handling code

The return value for the Observables subscribe method is a subscribe interface:

Observables subscribe方法的返回值是一个subscribe接口。

String[] letters = {"a", "b", "c", "d", "e", "f", "g"};
Observable<String> observable = Observable.from(letters);
observable.subscribe(
  i -> result += i,  //OnNext
  Throwable::printStackTrace, //OnError
  () -> result += "_Completed" //OnCompleted
);
assertTrue(result.equals("abcdefg_Completed"));

5. Observable Transformations and Conditional Operators

5.可观察的变换和条件运算符

5.1. Map

5.1.地图

The map operator transforms items emitted by an Observable by applying a function to each item.

map操作符通过对每个项目应用一个函数来转换由Observable发出的项目。

Let’s assume there is a declared array of strings that contains some letters from the alphabet and we want to print them in capital mode:

让我们假设有一个已声明的字符串数组，其中包含一些字母，我们想以大写模式打印它们。

Observable.from(letters)
  .map(String::toUpperCase)
  .subscribe(letter -> result += letter);
assertTrue(result.equals("ABCDEFG"));

The flatMap can be used to flatten Observables whenever we end up with nested Observables.

flatMap可以用来平坦Observables，只要我们最终有嵌套的Observables。

More details about the difference between map and flatMap can be found here.

关于map和flatMap之间的区别的更多细节可以在这里找到。

Assuming we have a method that returns an Observable<String> from a list of strings. Now we’ll be printing for each string from a new Observable the list of titles based on what Subscriber sees:

假设我们有一个方法，从一个字符串列表中返回一个Observable<String>。现在我们将从一个新的Observable中为每个字符串打印基于Subscriber看到的标题列表。

Observable<String> getTitle() {
    return Observable.from(titleList);
}
Observable.just("book1", "book2")
  .flatMap(s -> getTitle())
  .subscribe(l -> result += l);

assertTrue(result.equals("titletitle"));

5.2. Scan

5.2.扫描

The scan operator applies a function to each item emitted by an Observable sequentially and emits each successive value.

扫描操作符a对Observable发出的每个项目依次应用一个函数，并发出每个连续的值。

It allows us to carry forward state from event to event:

它允许我们将状态从一个事件延续到另一个事件。

String[] letters = {"a", "b", "c"};
Observable.from(letters)
  .scan(new StringBuilder(), StringBuilder::append)
  .subscribe(total -> result += total.toString());
assertTrue(result.equals("aababc"));

5.3. GroupBy

5.3.GroupBy

Group by operator allows us to classify the events in the input Observable into output categories.

Group by 操作器允许我们将输入Observable中的事件分类到输出类别。

Let’s assume that we created an array of integers from 0 to 10, then apply group by that will divide them into the categories even and odd:

让我们假设我们创建了一个从0到10的整数数组，然后应用group by，将它们分成even和odd两类。

Observable.from(numbers)
  .groupBy(i -> 0 == (i % 2) ? "EVEN" : "ODD")
  .subscribe(group ->
    group.subscribe((number) -> {
        if (group.getKey().toString().equals("EVEN")) {
            EVEN[0] += number;
        } else {
            ODD[0] += number;
        }
    })
  );
assertTrue(EVEN[0].equals("0246810"));
assertTrue(ODD[0].equals("13579"));

5.4. Filter

5.4.过滤器

The operator filter emits only those items from an observable that pass a predicate test.

操作符filter只接收observable中通过predicate测试的那些项目。

So let’s filter in an integer array for the odd numbers:

因此，让我们用一个整数数组来过滤奇数。

Observable.from(numbers)
  .filter(i -> (i % 2 == 1))
  .subscribe(i -> result += i);
 
assertTrue(result.equals("13579"));

5.5. Conditional Operators

5.5.条件性操作符

DefaultIfEmpty emits item from the source Observable, or a default item if the source Observable is empty:

DefaultIfEmpty从源Observable发出项目，如果源Observable是空的，则发出一个默认项目。

Observable.empty()
  .defaultIfEmpty("Observable is empty")
  .subscribe(s -> result += s);
 
assertTrue(result.equals("Observable is empty"));

The following code emits the first letter of the alphabet ‘a’ because the array letters is not empty and this is what it contains in the first position:

下面的代码发出了字母表的第一个字母’a’ ，因为数组letters 不是空的，这就是它在第一个位置所包含的内容。

Observable.from(letters)
  .defaultIfEmpty("Observable is empty")
  .first()
  .subscribe(s -> result += s);
 
assertTrue(result.equals("a"));

TakeWhile operator discards items emitted by an Observable after a specified condition becomes false:

TakeWhile操作者在指定的条件变为false后，丢弃由Observable发出的项目。

Observable.from(numbers)
  .takeWhile(i -> i < 5)
  .subscribe(s -> sum[0] += s);
 
assertTrue(sum[0] == 10);

Of course, there more others operators that could cover our needs like Contain, SkipWhile, SkipUntil, TakeUntil, etc.

当然，还有更多其他的操作符可以满足我们的需求，比如Contain, SkipWhile, SkipUntil, TakeUntil, 等等。

6. Connectable Observables

6.可连接的可观察变量

A ConnectableObservable resembles an ordinary Observable, except that it doesn’t begin emitting items when it is subscribed to, but only when the connect operator is applied to it.

一个ConnectableObservable类似于一个普通的Observable，除了它在被订阅时不开始发射项目，而只是在connect操作符被应用到它时才开始发射。

In this way, we can wait for all intended observers to subscribe to the Observable before the Observable begins emitting items:

通过这种方式，我们可以在Observable开始发射项目之前，等待所有预定的观察者订阅到Observable。

String[] result = {""};
ConnectableObservable<Long> connectable
  = Observable.interval(200, TimeUnit.MILLISECONDS).publish();
connectable.subscribe(i -> result[0] += i);
assertFalse(result[0].equals("01"));

connectable.connect();
Thread.sleep(500);
 
assertTrue(result[0].equals("01"));

7. Single

7.单一

Single is like an Observable who, instead of emitting a series of values, emits one value or an error notification.

Single就像一个Observable，它不是发出一系列的值，而是发出一个值或一个错误通知。

With this source of data, we can only use two methods to subscribe:

有了这个数据源，我们只能用两种方法来订阅。

  • OnSuccess returns a Single that also calls a method we specify
  • OnError also returns a Single that immediately notifies subscribers of an error
String[] result = {""};
Single<String> single = Observable.just("Hello")
  .toSingle()
  .doOnSuccess(i -> result[0] += i)
  .doOnError(error -> {
      throw new RuntimeException(error.getMessage());
  });
single.subscribe();
 
assertTrue(result[0].equals("Hello"));

8. Subjects

8.主题

A Subject is simultaneously two elements, a subscriber and an observable. As a subscriber, a subject can be used to publish the events coming from more than one observable.

一个主体同时是两个元素，一个订阅者和一个可观察者。作为一个订阅者，主体可以被用来发布来自多个观察者的事件。

And because it’s also observable, the events from multiple subscribers can be reemitted as its events to anyone observing it.

而且由于它也是可观察的，来自多个订阅者的事件可以作为它的事件重新释放给观察它的任何人。

In the next example, we’ll look at how the observers will be able to see the events that occur after they subscribe:

在下一个例子中，我们将看看观察员如何能够看到他们订阅后发生的事件。

Integer subscriber1 = 0;
Integer subscriber2 = 0;
Observer<Integer> getFirstObserver() {
    return new Observer<Integer>() {
        @Override
        public void onNext(Integer value) {
           subscriber1 += value;
        }

        @Override
        public void onError(Throwable e) {
            System.out.println("error");
        }

        @Override
        public void onCompleted() {
            System.out.println("Subscriber1 completed");
        }
    };
}

Observer<Integer> getSecondObserver() {
    return new Observer<Integer>() {
        @Override
        public void onNext(Integer value) {
            subscriber2 += value;
        }

        @Override
        public void onError(Throwable e) {
            System.out.println("error");
        }

        @Override
        public void onCompleted() {
            System.out.println("Subscriber2 completed");
        }
    };
}

PublishSubject<Integer> subject = PublishSubject.create(); 
subject.subscribe(getFirstObserver()); 
subject.onNext(1); 
subject.onNext(2); 
subject.onNext(3); 
subject.subscribe(getSecondObserver()); 
subject.onNext(4); 
subject.onCompleted();
 
assertTrue(subscriber1 + subscriber2 == 14)

9. Resource Management

9.资源管理

Using operation allows us to associate resources, such as a JDBC database connection, a network connection, or open files to our observables.

使用操作允许我们将资源，如JDBC数据库连接、网络连接或打开的文件与我们的观察变量相关联。

Here we’re presenting in commentaries the steps we need to do to achieve this goal and also an example of implementation:

在这里，我们以评论的形式介绍了为实现这一目标所需要做的步骤，也是一个实施的例子。

String[] result = {""};
Observable<Character> values = Observable.using(
  () -> "MyResource",
  r -> {
      return Observable.create(o -> {
          for (Character c : r.toCharArray()) {
              o.onNext(c);
          }
          o.onCompleted();
      });
  },
  r -> System.out.println("Disposed: " + r)
);
values.subscribe(
  v -> result[0] += v,
  e -> result[0] += e
);
assertTrue(result[0].equals("MyResource"));

10. Conclusion

10.结论

In this article, we have talked how to use RxJava library and also how to explore its most important features.

在这篇文章中，我们谈到了如何使用RxJava库以及如何探索其最重要的功能。

The full source code for the project including all the code samples used here can be found over on Github.

该项目的全部源代码，包括这里使用的所有代码样本，都可以在Github上找到over。