Java 类io.reactivex.internal.queue.SpscLinkedArrayQueue 实例源码

ValveMainSubscriber(Subscriber<? super T> actual, int bufferSize, boolean defaultOpen) {
    this.actual = actual;
    this.queue = new SpscLinkedArrayQueue<T>(bufferSize);
    this.gate = defaultOpen;
    this.other = new OtherSubscriber();
    this.requested = new AtomicLong();
    this.error = new AtomicThrowable();
    this.s = new AtomicReference<Subscription>();
}

SpanoutSubscriber(Subscriber<? super T> actual, long initialSpan, long betweenSpan,
        Worker worker, boolean delayError, int bufferSize) {
    this.actual = actual;
    this.initialSpan = initialSpan;
    this.betweenSpan = betweenSpan;
    this.worker = worker;
    this.delayError = delayError;
    this.lastEvent = -1L;
    this.queue = new SpscLinkedArrayQueue<T>(bufferSize);
}

ExpandBreadthSubscriber(Subscriber<? super T> actual,
        Function<? super T, ? extends Publisher<? extends T>> expander, int capacityHint, boolean delayErrors) {
    this.actual = actual;
    this.expander = expander;
    this.wip = new AtomicInteger();
    this.queue = new SpscLinkedArrayQueue<Publisher<? extends T>>(capacityHint);
    this.errors = new AtomicThrowable();
    this.delayErrors = delayErrors;
}

Chain(Function<? super Flowable<T>, ? extends Flowable<T>> transform,
        DestinationSerializedSubject<T> destination, long maxIterations, int maxChained,
        Function<Observable<T>, ? extends Observable<?>> test) {
    this.transform = transform;
    this.destination = destination;
    this.maxIterations = maxIterations;
    this.maxChained = maxChained;
    this.test = test;
    this.queue = new SpscLinkedArrayQueue<Event<T>>(16);
}

public void spsc1() {
    SpscLinkedArrayQueue<Integer> q = new SpscLinkedArrayQueue<>(capacity);
    int n = times;
    for (int i = 0; i < n; i++) {
        q.offer(0);
        q.poll();
    }
}

public void spscFill1() {
    SpscLinkedArrayQueue<Integer> q = new SpscLinkedArrayQueue<>(capacity);
    int n = times;
    for (int i = 0; i < n; i++) {
        q.offer(0);
    }
    for (int i = 0; i < n; i++) {
        q.poll();
    }
}