如何确保RXJava Observable仅发出且恰好1个项目

Observable.just(item1, item2, item3)
    .take(1)
    .subscribe(new Observer<Item>() {
        @Override
        public void onSubscribe(Disposable d) {
            // 订阅时的操作
        }

        @Override
        public void onNext(Item item) {
            // 处理接收到的项目
        }

        @Override
        public void onError(Throwable e) {
            // 处理错误
        }

        @Override
        public void onComplete() {
            // 处理完成
        }
    });

Observable.just(item1, item2, item3)
    .single()
    .subscribe(new SingleObserver<Item>() {
        @Override
        public void onSubscribe(Disposable d) {
            // 订阅时的操作
        }

        @Override
        public void onSuccess(Item item) {
            // 处理接收到的项目
        }

        @Override
        public void onError(Throwable e) {
            // 处理错误
        }
    });

Observable.just(item1, item2, item3)
    .filter(new Predicate<Item>() {
        @Override
        public boolean test(Item item) throws Exception {
            // 设置条件，只保留满足条件的项目
            return item.getId() == desiredId;
        }
    })
    .subscribe(new Observer<Item>() {
        @Override
        public void onSubscribe(Disposable d) {
            // 订阅时的操作
        }

        @Override
        public void onNext(Item item) {
            // 处理接收到的项目
        }

        @Override
        public void onError(Throwable e) {
            // 处理错误
        }

        @Override
        public void onComplete() {
            // 处理完成
        }
    });