如何在RxJava merge()中限制活动流的数量？

import io.reactivex.Observable;
import io.reactivex.schedulers.Schedulers;

import java.util.concurrent.Semaphore;

public class MergeWithConcurrencyLimitExample {
    public static void main(String[] args) {
        int concurrencyLimit = 3; // 同时活动的流数量限制为3

        Semaphore semaphore = new Semaphore(concurrencyLimit);

        Observable<Integer> source1 = Observable.range(1, 10);
        Observable<Integer> source2 = Observable.range(11, 10);
        Observable<Integer> source3 = Observable.range(21, 10);

        Observable.merge(
                source1.flatMap(item -> processItem(item, semaphore)),
                source2.flatMap(item -> processItem(item, semaphore)),
                source3.flatMap(item -> processItem(item, semaphore))
        )
                .subscribe(System.out::println);

        // 等待所有流处理完成
        try {
            semaphore.acquire(concurrencyLimit);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private static Observable<Integer> processItem(int item, Semaphore semaphore) {
        return Observable.just(item)
                .subscribeOn(Schedulers.io())
                .doOnSubscribe(disposable -> semaphore.acquireUninterruptibly()) // 获取信号量
                .doFinally(() -> semaphore.release()); // 释放信号量
    }
}