TopicProcessor深度解析：多订阅者场景下的高效数据分发机制¶

// 环形队列核心结构
private final RingBuffer<T> ringBuffer;
private final AtomicInteger cursor = new AtomicInteger();
private final CopyOnWriteArrayList<Subscription> subscriptions = new CopyOnWriteArrayList<>();

// onNext方法实现
public void onNext(T t) {
    if (t == null) {
        onError(new NullPointerException("onNext called with null"));
        return;
    }

    // 将元素写入环形队列
    long sequence = ringBuffer.next();
    try {
        ringBuffer.get(sequence).setValue(t);
    } finally {
        ringBuffer.publish(sequence);
    }

    // 通知所有订阅者消费
    for (Subscription s : subscriptions) {
        s.request(1);
    }
}

// subscribe方法实现
public void subscribe(Subscriber<? super T> subscriber) {
    Subscription subscription = new TopicSubscription(subscriber, ringBuffer);
    subscriptions.add(subscription);

    // 初始化订阅者消费位置
    subscription.setInitialCursor(cursor.get());

    subscriber.onSubscribe(subscription);
}

// TopicSubscription核心逻辑
public void onNext(T item) {
    long expectedCursor = cursor.get();
    if (expectedCursor >= item.getSequence()) {
        // 已处理过该数据
        return;
    }

    // 更新本地消费位置
    cursor.compareAndSet(expectedCursor, item.getSequence());
    actual.onNext(item.getValue());
}

// 日志生产者
Flux.range(1, 10000)
    .map(i -> "LogEntry-" + i)
    .publishOn(Schedulers.parallel())
    .subscribe(topicProcessor);

// 订阅者1（过滤ERROR日志）
topicProcessor
    .filter(log -> log.contains("ERROR"))
    .subscribe(System.out::println);

// 订阅者2（统计日志数量）
topicProcessor
    .buffer(Duration.ofSeconds(1))
    .subscribe(batch -> 
        System.out.println("Received " + batch.size() + " logs"));

topicProcessor.doOnNext(item -> 
    System.out.println("Buffer occupancy: " + item.getSequence()))
.subscribe();

topicProcessor
    .transform(event -> {
        long start = System.nanoTime();
        return event;
    })
    .doOnNext(event -> 
        System.out.println("Processing latency: " + 
            (System.nanoTime() - start) + " ns"))
    .subscribe();