本文整理了Java中rx.Observable.onBackpressureBuffer()方法的一些代码示例，展示了Observable.onBackpressureBuffer()的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。Observable.onBackpressureBuffer()方法的具体详情如下：
包路径：rx.Observable
类名称：Observable
方法名：onBackpressureBuffer

Observable.onBackpressureBuffer介绍

[英]Instructs an Observable that is emitting items faster than its observer can consume them to buffer these items indefinitely until they can be emitted.

Scheduler: onBackpressureBuffer does not operate by default on a particular Scheduler.
[中]指示发射项目的速度快于其观察者使用项目的速度的可观察者无限期地缓冲这些项目，直到它们可以发射。
调度程序：onBackpressureBuffer默认情况下不会在特定调度程序上运行。

代码示例

代码示例来源：origin: Rukey7/MvpApp

/**
 * 一个默认的订阅方法
 *
 * @param type
 * @param next
 * @param error
 * @param <T>
 * @return
 */
public <T> Subscription doSubscribe(Class<T> type, Action1<T> next, Action1<Throwable> error) {
  return toObservable(type)
      // 加上背压处理，不然有些地方会有异常，关于背压参考这里：https://gold.xitu.io/post/582d413c8ac24700619cceed
      .onBackpressureBuffer()
      .subscribeOn(Schedulers.io())
      .observeOn(AndroidSchedulers.mainThread())
      .subscribe(next, error);
}

代码示例来源：origin: mesosphere/mesos-rxjava

/**
 * Instructs the HTTP byte[] stream to be composed with reactive pull backpressure such that
 * a burst of incoming Mesos messages is handled by an unbounded buffer rather than a
 * MissingBackpressureException.
 *
 * As an example, this may be necessary for Mesos schedulers that launch large numbers
 * of tasks at a time and then request reconciliation.
 *
 * @return this builder (allowing for further chained calls)
 * @see <a href="http://reactivex.io/documentation/operators/backpressure.html">ReactiveX operators documentation: backpressure operators</a>
 */
@NotNull
public MesosClientBuilder<Send, Receive> onBackpressureBuffer(
) {
  this.backpressureTransformer = observable -> observable.onBackpressureBuffer();
  return this;
}

代码示例来源：origin: com.mesosphere.mesos.rx.java/mesos-rxjava-client

/**
 * Instructs the HTTP byte[] stream to be composed with reactive pull backpressure such that
 * a burst of incoming Mesos messages is handled by an unbounded buffer rather than a
 * MissingBackpressureException.
 *
 * As an example, this may be necessary for Mesos schedulers that launch large numbers
 * of tasks at a time and then request reconciliation.
 *
 * @return this builder (allowing for further chained calls)
 * @see <a href="http://reactivex.io/documentation/operators/backpressure.html">ReactiveX operators documentation: backpressure operators</a>
 */
@NotNull
public MesosClientBuilder<Send, Receive> onBackpressureBuffer(
) {
  this.backpressureTransformer = observable -> observable.onBackpressureBuffer();
  return this;
}

代码示例来源：origin: com.mesosphere.mesos.rx.java/mesos-rxjava-client

/**
 * Instructs the HTTP byte[] stream to be composed with reactive pull backpressure such that
 * a burst of incoming Mesos messages is handled by a bounded buffer rather than a
 * MissingBackpressureException. If the buffer is overflown, a {@link java.nio.BufferOverflowException}
 * is thrown.
 *
 * As an example, this may be necessary for Mesos schedulers that launch large numbers
 * of tasks at a time and then request reconciliation.
 *
 * @param capacity number of slots available in the buffer.
 * @return this builder (allowing for further chained calls)
 * @see <a href="http://reactivex.io/documentation/operators/backpressure.html">ReactiveX operators documentation: backpressure operators</a>
 */
@NotNull
public MesosClientBuilder<Send, Receive> onBackpressureBuffer(
  final long capacity
) {
  this.backpressureTransformer = observable -> observable.onBackpressureBuffer(capacity);
  return this;
}

代码示例来源：origin: HelloChenJinJun/TestChat

public <T> Subscription registerEvent(Class<T> type, Action1<T> action1, Action1<Throwable> error) {
    return getObservable(type)
        .onBackpressureBuffer().subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(action1, error);
}

代码示例来源：origin: HelloChenJinJun/TestChat

public <T> Subscription registerEvent(Class<T> type, Action1<T> action1, Action1<Throwable> error) {
    return getObservable(type)
        .onBackpressureBuffer().subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(action1, error);
}

代码示例来源：origin: SmartDengg/RxBlur

@CheckResult @NonNull
public static Observable<Bitmap> animatorBlur(@NonNull Context context, @DrawableRes int resId, int radius,
  long duration) {
 return Observable.create(new AnimationBlurOnSubscribe(context, resId, radius, duration)).onBackpressureBuffer();
}

代码示例来源：origin: davidmoten/rxjava-extras

private static <Out> Observable<Notification<Out>> applyBackpressure(
    Observable<Notification<Out>> o, final BackpressureStrategy backpressureStrategy) {
  if (backpressureStrategy == BackpressureStrategy.BUFFER)
    return o.onBackpressureBuffer();
  else if (backpressureStrategy == BackpressureStrategy.DROP)
    return o.onBackpressureDrop();
  else if (backpressureStrategy == BackpressureStrategy.LATEST)
    return o.onBackpressureLatest();
  else
    throw new IllegalArgumentException(
        "backpressure strategy not supported: " + backpressureStrategy);
}

代码示例来源：origin: com.github.davidmoten/rxjava-extras

private static <Out> Observable<Notification<Out>> applyBackpressure(
    Observable<Notification<Out>> o, final BackpressureStrategy backpressureStrategy) {
  if (backpressureStrategy == BackpressureStrategy.BUFFER)
    return o.onBackpressureBuffer();
  else if (backpressureStrategy == BackpressureStrategy.DROP)
    return o.onBackpressureDrop();
  else if (backpressureStrategy == BackpressureStrategy.LATEST)
    return o.onBackpressureLatest();
  else
    throw new IllegalArgumentException(
        "backpressure strategy not supported: " + backpressureStrategy);
}

代码示例来源：origin: marcoRS/rxjava-essentials

private void loadList(List<AppInfo> apps) {
 mRecyclerView.setVisibility(View.VISIBLE);
 getObservableApps(apps).onBackpressureBuffer()
   .subscribeOn(Schedulers.computation())
   .observeOn(AndroidSchedulers.mainThread())
   .subscribe(new Observer<AppInfo>() {
    @Override public void onCompleted() {
     mSwipeRefreshLayout.setRefreshing(false);
     Toast.makeText(getActivity(), "Here is the list!", Toast.LENGTH_LONG).show();
    }
    @Override public void onError(Throwable e) {
     Toast.makeText(getActivity(), "Something went wrong!", Toast.LENGTH_SHORT).show();
     if (mSwipeRefreshLayout != null) {
      mSwipeRefreshLayout.setRefreshing(false);
     }
    }
    @Override public void onNext(AppInfo appInfo) {
     mAddedApps.add(appInfo);
     mAdapter.addApplication(mAddedApps.size() - 1, appInfo);
    }
   });
}

代码示例来源：origin: easemob/kefu-android-demo

/**
 * 订阅文件下载进度
 */
private void subscribeLoadProgress() {
  rxSubscriptions.add(RxBus.getInstance()
      .toObservable(FileLoadEvent.class)
      .onBackpressureBuffer()
      .subscribeOn(Schedulers.io())
      .observeOn(AndroidSchedulers.mainThread())
      .subscribe(new Action1<FileLoadEvent>() {
        @Override
        public void call(FileLoadEvent fileLoadEvent) {
          progress(fileLoadEvent.getProgress(), fileLoadEvent.getTotal());
        }
      }));
}

代码示例来源：origin: shanyao0/DownLoadManager

/**
 * 订阅文件下载进度
 */
private void subscribeLoadProgress() {
  rxSubscriptions.add(RxBus.getDefault()
      .toObservable(FileLoadingBean.class)
      .onBackpressureBuffer()
      .subscribeOn(Schedulers.io())
      .observeOn(AndroidSchedulers.mainThread())
      .subscribe(new Action1<FileLoadingBean>() {
        @Override
        public void call(FileLoadingBean fileLoadEvent) {
          onLoading(fileLoadEvent.getProgress(), fileLoadEvent.getTotal());
        }
      }));
}
/**

代码示例来源：origin: org.springframework.cloud/spring-cloud-netflix-hystrix

public Function<HystrixObservableCommand<T>, Observable<T>> getObservableFunction() {
  Function<HystrixObservableCommand<T>, Observable<T>> observableFunc;
  if (this.toObservable != null) {
    observableFunc = this.toObservable;
  } else if (this.eager) {
    observableFunc = cmd -> cmd.observe();
  } else { // apply a default onBackpressureBuffer if not eager
    observableFunc = cmd -> cmd.toObservable().onBackpressureBuffer();
  }
  return observableFunc;
}

代码示例来源：origin: dswarm/dswarm

protected ConnectableObservable<GDMModel> transformResultModel(final Observable<org.dswarm.persistence.model.internal.Model> model) {
  final AtomicInteger resultCounter = new AtomicInteger(0);
  return model
      .onBackpressureBuffer(10000)
      .doOnSubscribe(() -> GDMModelTransformationFlow.LOG.debug("subscribed to results observable in transformation engine"))
      .doOnNext(resultObj -> {
        resultCounter.incrementAndGet();
        if (resultCounter.get() == 1) {
          GDMModelTransformationFlow.LOG.debug("received first result in transformation engine");
        }
      })
      .doOnCompleted(() -> GDMModelTransformationFlow.LOG.debug("received '{}' results in transformation engine overall", resultCounter.get()))
      .cast(org.dswarm.persistence.model.internal.gdm.GDMModel.class)
      .onBackpressureBuffer(10000)
      .publish();
}

代码示例来源：origin: techery/janet

@Override public Observable<ActionState<A>> call() {
    return pipeline.asObservable()
        .onBackpressureBuffer()
        .map(new Func1<ActionPair, ActionState>() {
          @Override public ActionState call(ActionPair pair) {
            return pair.state;
          }
        })
        .filter(new Func1<ActionState, Boolean>() {
          @Override public Boolean call(ActionState actionState) {
            return actionClass.isInstance(actionState.action);
          }
        }).compose(new CastToState<A>());
  }
}, new Action1<A>() {

代码示例来源：origin: dswarm/dswarm

@Override
public Observable<JsonNode> generate(final Observable<GDMModel> recordGDM,
                   final OutputStream outputStream) throws XMLStreamException {
  final StreamRDF writer = StreamRDFWriter.getWriterStream(outputStream, rdfSerializationFormat);
  writer.start();
  final ConcurrentHashMap<String, org.apache.jena.graph.Node> resourceNodeCache = new ConcurrentHashMap<>();
  final ConcurrentHashMap<String, org.apache.jena.graph.Node> predicateCache = new ConcurrentHashMap<>();
  return recordGDM
      .doOnSubscribe(() -> LOG.debug("subscribed to RDF export; will return data as '{}'", mediaType.toString()))
      .onBackpressureBuffer(10000)
      .map(recordGDMModel -> processRecordGDMModel(writer, resourceNodeCache, predicateCache, recordGDMModel))
      .map(org.dswarm.persistence.model.internal.Model::toGDMCompactJSON)
      .flatMapIterable(nodes -> {
        final ArrayList<JsonNode> nodeList = new ArrayList<>();
        Iterators.addAll(nodeList, nodes.elements());
        return nodeList;
      })
      .doOnCompleted(writer::finish)
      .doOnCompleted(() -> LOG.debug("finished RDF export; return data as '{}'", mediaType.toString()));
}

代码示例来源：origin: davidmoten/rxjava-extras

@Override
  public Observable<T> call() {
    final OperatorPassThroughAdjustedRequest<T> op = new OperatorPassThroughAdjustedRequest<T>();
    return o.lift(op).onBackpressureBuffer().doOnRequest(new Action1<Long>() {
      @Override
      public void call(Long n) {
        op.requestMore(n);
      }
    });
  }
});

代码示例来源：origin: com.github.davidmoten/rxjava-extras

@Override
  public Observable<T> call() {
    final OperatorPassThroughAdjustedRequest<T> op = new OperatorPassThroughAdjustedRequest<T>();
    return o.lift(op).onBackpressureBuffer().doOnRequest(new Action1<Long>() {
      @Override
      public void call(Long n) {
        op.requestMore(n);
      }
    });
  }
});

代码示例来源：origin: dswarm/dswarm

protected ConnectableObservable<org.dswarm.persistence.model.internal.Model> doPostProcessingOfResultModel(final GDMModelReceiver writer, final Scheduler scheduler) {
  final ConnectableObservable<GDMModel> modelConnectableObservable = writer.getObservable()
      .observeOn(scheduler)
      .onBackpressureBuffer(10000)
      .publish();
  final ConnectableObservable<org.dswarm.persistence.model.internal.Model> model = doPostProcessingOfResultModel(modelConnectableObservable);
  modelConnectableObservable.connect();
  return model;
}

代码示例来源：origin: avluis/Hentoid

void process(File rootDir) {
  if (rootDir.canRead()) {
    cancelPrevOp();
    updateDirList(rootDir);
    Observable<File> observable = new ListDirObservable().create(rootDir);
    Observer<File> observer = new ListDirObserver(dirTree, bus);
    subscription = observable.subscribeOn(Schedulers.io())
        .onBackpressureBuffer()
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(observer);
  } else {
    Timber.d("Failed to process directory list.");
    bus.post(new OpFailedEvent());
  }
}