RxJava阀门使用案例

Observable<UserInfo> userInfoObservable = Observable.create(new ObservableOnSubscribe<UserInfo>() {
    @Override
    public void subscribe(ObservableEmitter<UserInfo> emitter) throws Exception {
        // 从数据源获取用户信息并发射
        List<UserInfo> userInfoList = DataSource.getUserInfoList();
        for (UserInfo userInfo : userInfoList) {
            emitter.onNext(userInfo);
        }
        emitter.onComplete();
    }
})
.filter(new Predicate<UserInfo>() {
    @Override
    public boolean test(UserInfo userInfo) throws Exception {
        // 过滤年龄小于18岁的用户信息
        return userInfo.getAge() >= 18;
    }
})
.throttleFirst(100, TimeUnit.MILLISECONDS, Schedulers.computation())
.subscribe(new Consumer<UserInfo>() {
    @Override
    public void accept(UserInfo userInfo) throws Exception {
        // 处理筛选后的用户信息
        System.out.println(userInfo.getName() + "，年龄：" + userInfo.getAge());
    }
});