src/main/java/io/reactivex/internal/operators/observable/ObservableCache.java
void add(CacheDisposable<T> consumer) {
        for (;;) {
            CacheDisposable<T>[] current = observers.get();
            if (current == TERMINATED) {
                return;
            }
            int n = current.length;

            @SuppressWarnings("unchecked")
            CacheDisposable<T>[] next = new CacheDisposable[n + 1];
            System.arraycopy(current, 0, next, 0, n);
            next[n] = consumer;

            if (observers.compareAndSet(current, next)) {
                return;
            }
        }
    }
void add(CacheDisposable<T> consumer) {
        for (;;) {
            CacheDisposable<T>[] current = observers.get();
            if (current == TERMINATED) {
                return;
            }
            int n = current.length;

            @SuppressWarnings("unchecked")
            CacheDisposable<T>[] next = new CacheDisposable[n + 1];
            System.arraycopy(current, 0, next, 0, n);
            next[n] = consumer;

            if (observers.compareAndSet(current, next)) {
                return;
            }
        }
    }
Atomically adds the consumer to the {@link #observers} copy-on-write array
if the source has not yet terminated.
@param consumer the consumer to add