容易被忽略的知识点RxJava操作符的线程安全，kotlin源码

val publishSubject = PublishSubject.create()

val actuallyReceived = AtomicInteger()

publishSubject.take(3).subscribe {

actuallyReceived.incrementAndGet()

}

val latch = CountDownLatch(numberOfThreads)

var threads = listOf()

(0…numberOfThreads).forEach {

threads = thread(start = false) {

publishSubject.onNext(it)

latch.countDown()

}

}

threads.forEach { it.start() }

latch.await()

check(actuallyReceived.get() == 3)

}

}

执行上面代码，由于take的结果不符合预期，总是会异常退出

看一下take的源码：

public final class ObservableTake extends AbstractObservableWithUpstream<T, T> {

final long limit;

public ObservableTake(ObservableSource source,

这篇好文章是转载于：学新通技术网