在并发情况下,不推荐使用通常的Subject对象,而是推荐使用SerializedSubject,并发时只允许一个线程调用onnext等方法!
官方说明:

1
When you use an ordinary Subject as a Subscriber, you must take care not to call its Subscriber.onNext method (or its other on methods) from multiple threads, as this could lead to non-serialized calls, which violates the Observable contract and creates an ambiguity in the resulting Subject.

大致意思是当我们使用普通的Subject,必须要注意不要在多线程情况下调用onNext 方法,这样是违反了Observable 协议并且会导致执行结果返回带有有歧义的值(线程并发导致返回值混淆了)!

1
2
3
To protect a Subject from this danger, you can convert it into a SerializedSubject with code like the following: 

mySafeSubject = new SerializedSubject( myUnsafeSubject );

官方文档说的很明白了,只需要使用SerializedSubject封装原来的
Subject即可!!

测试demo:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
public class MultiThread {
public static void main(String[] args) throws InterruptedException {

final PublishSubject<Integer> subject = PublishSubject.create();

subject.subscribe(new Action1<Integer>() {

@Override
public void call(Integer t) {
System.out.println("======onnext===>value:" + t + ",threadId:" + Thread.currentThread().getId());
}

});

final SerializedSubject<Integer, Integer> ser = new SerializedSubject<Integer, Integer>(subject);

for (int i = 0; i < 20; i++) {
final int value = i;
new Thread() {
public void run() {
ser.onNext((int) (value * 10000 + Thread.currentThread().getId()));
};
}.start();
}

Thread.sleep(2000);
//
// for (int i = 11; i < 20; i++) {
// final int value = i;
// new Thread() {
// public void run() {
// subject.onNext(value);
// };
// }.start();
// }

}
}

执行结果:

======onnext===>value:10,threadId:10
======onnext===>value:10011,threadId:10
======onnext===>value:50015,threadId:10
======onnext===>value:40014,threadId:10
======onnext===>value:30013,threadId:10
======onnext===>value:20012,threadId:10
======onnext===>value:70017,threadId:10
======onnext===>value:60016,threadId:10
======onnext===>value:80018,threadId:10
======onnext===>value:100020,threadId:10
======onnext===>value:90019,threadId:10
======onnext===>value:110021,threadId:21
======onnext===>value:130023,threadId:23
======onnext===>value:120022,threadId:23
======onnext===>value:160026,threadId:26
======onnext===>value:150025,threadId:25
======onnext===>value:140024,threadId:25
======onnext===>value:170027,threadId:25
======onnext===>value:180028,threadId:25
======onnext===>value:190029,threadId:25

上面的结果有点晕了,为什么不是在一个线程上呢?和我之前以为的SerializedSubject时将值放在一个线程上然后处理的想法有些出入了!

源码面前了无秘密,SerializedSubject跟进去看看

1
2
3
4
5
6
7
8
9
10
11
12
public SerializedSubject(final Subject<T, R> actual) {
super(new OnSubscribe<R>() {

@Override
public void call(Subscriber<? super R> child) {
actual.unsafeSubscribe(child);
}

});
this.actual = actual;
this.observer = new SerializedObserver<T>(actual);
}

其实SerializedSubject的处理是交给了SerializedObserver,继续跟进到SerializedObserver,类注释:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
/**
* Enforces single-threaded, serialized, ordered execution of {@link #onNext}, {@link #onCompleted}, and
* {@link #onError}.
* <p>
* When multiple threads are emitting and/or notifying they will be serialized by:
* </p><ul>
* <li>Allowing only one thread at a time to emit</li>
* <li>Adding notifications to a queue if another thread is already emitting</li>
* <li>Not holding any locks or blocking any threads while emitting</li>
* </ul>
*
* @param <T>
* the type of items expected to be observed by the {@code Observer}
*/

这里一看就明白了,他是只保证同时只有一个线程调用 {@link #onNext}, {@link #onCompleted}, and{@link #onError}.方法,并不是将所有emit的值放到一个线程上然后处理,这就解释了为什么执行结果不是全部在一个线程上的原因 了!

再看看源码再onnext方法:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
  @Override
public void onNext(T t) {
FastList list;

//同步锁
synchronized (this) {
if (terminated) {
return;
}
if (emitting) {
if (queue == null) {
queue = new FastList();
}
queue.add(t != null ? t : NULL_SENTINEL);
// another thread is emitting so we add to the queue and return
return;
}
// we can emit
emitting = true;
// reference to the list to drain before emitting our value
list = queue;
queue = null;
}

// we only get here if we won the right to emit, otherwise we returned in the if(emitting) block above
boolean skipFinal = false;
try {
int iter = MAX_DRAIN_ITERATION;
do {
drainQueue(list);
if (iter == MAX_DRAIN_ITERATION) {
// after the first draining we emit our own value
actual.onNext(t);
}
--iter;
if (iter > 0) {
synchronized (this) {
list = queue;
queue = null;
if (list == null) {
emitting = false;
skipFinal = true;
return;
}
}
}
} while (iter > 0);
} finally {
if (!skipFinal) {
synchronized (this) {
if (terminated) {
list = queue;
queue = null;
} else {
emitting = false;
list = null;
}
}
}
}

// this will only drain if terminated (done here outside of synchronized block)
drainQueue(list);
}

// another thread is emitting so we add to the queue and return
如果有其他线程正在处理,则将emit的值放到队列上,线程执行完毕后,会顺序emit队列上的值!!这样就保证了一次只会有一个线程调用!!!