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

Observable介绍

[英]The Observable class that implements the Reactive Pattern.

This class provides methods for subscribing to the Observable as well as delegate methods to the various Observers.

The documentation for this class makes use of marble diagrams. The following legend explains these diagrams:

For more information see the RxJava wiki
[中]实现反应模式的可观察类。
此类提供订阅可观察对象的方法，以及将方法委托给各种观察者。
这个类的文档使用大理石图表。以下图例解释了这些图表：
有关更多信息，请参见{$0$}

代码示例

代码示例来源：origin: HotBitmapGG/bilibili-android-client

private void getTags() {
  RetrofitHelper.getSearchAPI()
      .getHotSearchTags()
      .compose(bindToLifecycle())
      .map(HotSearchTag::getList)
      .subscribeOn(Schedulers.io())
      .observeOn(AndroidSchedulers.mainThread())
      .subscribe(listBeans -> {
        hotSearchTags.addAll(listBeans);
        initTagLayout();
      }, throwable -> {
      });
}

代码示例来源：origin: hidroh/materialistic

@Override
public void parse(String itemId, String url, Callback callback) {
  Observable.defer(() -> fromCache(itemId))
      .subscribeOn(mIoScheduler)
      .flatMap(content -> content != null ?
          Observable.just(content) : fromNetwork(itemId, url))
      .map(content -> AndroidUtils.TextUtils.equals(EMPTY_CONTENT, content) ? null : content)
      .observeOn(mMainThreadScheduler)
      .subscribe(callback::onResponse);
}

代码示例来源：origin: PipelineAI/pipeline

@Override
protected Observable<Boolean> construct() {
  return Observable.create(new OnSubscribe<Boolean>() {
    @Override
    public void call(Subscriber<? super Boolean> s) {
      s.onError(new RuntimeException("onError"));
    }
  }).subscribeOn(userScheduler);
}

代码示例来源：origin: Netflix/zuul

public Observable<String> makeSlowRequest() {
    return Observable.just("test").delay(500, TimeUnit.MILLISECONDS);
  }
}

代码示例来源：origin: greenrobot/greenDAO

@Override
  public Observable<T> call() {
    T result;
    try {
      result = callable.call();
    } catch (Exception e) {
      return Observable.error(e);
    }
    return Observable.just(result);
  }
});

代码示例来源：origin: PipelineAI/pipeline

@Override
  public Observable<Bucket> call() {
    return inputEventStream
        .observe()
        .window(bucketSizeInMs, TimeUnit.MILLISECONDS) //bucket it by the counter window so we can emit to the next operator in time chunks, not on every OnNext
        .flatMap(reduceBucketToSummary)                //for a given bucket, turn it into a long array containing counts of event types
        .startWith(emptyEventCountsToStart);           //start it with empty arrays to make consumer logic as generic as possible (windows are always full)
  }
});

代码示例来源：origin: hidroh/materialistic

@Override
public void login(String username, String password, boolean createAccount, Callback callback) {
  execute(postLogin(username, password, createAccount))
      .flatMap(response -> {
        if (response.code() == HttpURLConnection.HTTP_OK) {
          return Observable.error(new UserServices.Exception(parseLoginError(response)));
        }
        return Observable.just(response.code() == HttpURLConnection.HTTP_MOVED_TEMP);
      })
      .observeOn(AndroidSchedulers.mainThread())
      .subscribe(callback::onDone, callback::onError);
}

代码示例来源：origin: GitLqr/LQRWeChat

private void setQRCode(String content) {
  Observable.just(QRCodeEncoder.syncEncodeQRCode(content, UIUtils.dip2Px(100)))
      .subscribeOn(Schedulers.io())
      .observeOn(AndroidSchedulers.mainThread())
      .subscribe(bitmap -> mIvCard.setImageBitmap(bitmap), this::loadQRCardError);
}

代码示例来源：origin: amitshekhariitbhu/Fast-Android-Networking

public void downloadFile(View view) {
    subscription = getObservable()
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(getObserver());
  }
}

代码示例来源：origin: Netflix/conductor

@VisibleForTesting
OnSubscribe<Message> getOnSubscribe() {
  return subscriber -> {
    Observable<Long> interval = Observable.interval(pollTimeInMS, TimeUnit.MILLISECONDS);
    interval.flatMap((Long x)->{
      List<Message> msgs = receiveMessages();
      return Observable.from(msgs);
    }).subscribe(subscriber::onNext, subscriber::onError);
  };
}

代码示例来源：origin: greenrobot/greenDAO

private void updateNotes() {
  notesQuery.list()
      .observeOn(AndroidSchedulers.mainThread())
      .subscribe(new Action1<List<Note>>() {
        @Override
        public void call(List<Note> notes) {
          notesAdapter.setNotes(notes);
        }
      });
}

代码示例来源：origin: com.microsoft.azure/azure-mgmt-resources

@Override
  public Observable<ServiceResponse<Page<SubscriptionInner>>> call(ServiceResponse<Page<SubscriptionInner>> page) {
    String nextPageLink = page.body().nextPageLink();
    if (nextPageLink == null) {
      return Observable.just(page);
    }
    return Observable.just(page).concatWith(listNextWithServiceResponseAsync(nextPageLink));
  }
});

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

/**
 * 获取福利图片
 * @return
 */
public static Observable<WelfarePhotoInfo> getWelfarePhoto(int page) {
  return sWelfareService.getWelfarePhoto(page)
      .subscribeOn(Schedulers.io())
      .unsubscribeOn(Schedulers.io())
      .subscribeOn(AndroidSchedulers.mainThread())
      .observeOn(AndroidSchedulers.mainThread())
      .flatMap(_flatMapWelfarePhotos());
}

代码示例来源：origin: PipelineAI/pipeline

public void startCachingStreamValuesIfUnstarted() {
  if (rollingDistributionSubscription.get() == null) {
    //the stream is not yet started
    Subscription candidateSubscription = observe().subscribe(rollingDistribution);
    if (rollingDistributionSubscription.compareAndSet(null, candidateSubscription)) {
      //won the race to set the subscription
    } else {
      //lost the race to set the subscription, so we need to cancel this one
      candidateSubscription.unsubscribe();
    }
  }
}

代码示例来源：origin: PipelineAI/pipeline

@Override
protected Observable<Integer> construct() {
  return Observable.just(1, 2, 3)
      .concatWith(Observable.<Integer> error(new RuntimeException("forced error")))
      .subscribeOn(Schedulers.computation());
}

代码示例来源：origin: HotBitmapGG/bilibili-android-client

private void initRxBus() {
  RxBus.getInstance().toObserverable(Integer.class)
      .compose(bindToLifecycle())
      .observeOn(AndroidSchedulers.mainThread())
      .subscribe(this::switchPager);
}

代码示例来源：origin: PipelineAI/pipeline

@Test
public void noEvents() throws InterruptedException {
  CountDownLatch commandLatch = new CountDownLatch(1);
  CountDownLatch threadPoolLatch = new CountDownLatch(1);
  Subscriber<HystrixCommandCompletion> commandSubscriber = getLatchedSubscriber(commandLatch);
  readCommandStream.observe().take(1).subscribe(commandSubscriber);
  Subscriber<HystrixCommandCompletion> threadPoolSubscriber = getLatchedSubscriber(threadPoolLatch);
  readThreadPoolStream.observe().take(1).subscribe(threadPoolSubscriber);
  //no writes
  assertFalse(commandLatch.await(1000, TimeUnit.MILLISECONDS));
  assertFalse(threadPoolLatch.await(1000, TimeUnit.MILLISECONDS));
}

代码示例来源：origin: apache/usergrid

private List<MarkedEdge> createConnectionSearchEdges( final Entity testEntity, final GraphManager graphManager,
                           final int edgeCount ) {
  final List<MarkedEdge> connectionSearchEdges = Observable.range( 0, edgeCount ).flatMap( integer -> {
    //create our connection edge.
    final Id connectingId = createId( "connecting" );
    final Edge connectionEdge = CpNamingUtils.createConnectionEdge( connectingId, "likes", testEntity.getId() );
    return graphManager.writeEdge( connectionEdge ).subscribeOn( Schedulers.io() );
  }, 20).toList().toBlocking().last();
  assertEquals( "All edges saved", edgeCount, connectionSearchEdges.size() );
  return connectionSearchEdges;
}

代码示例来源：origin: PipelineAI/pipeline

@Override
protected Observable<String> construct() {
  executed = true;
  return Observable.just(value).delay(duration, TimeUnit.MILLISECONDS).subscribeOn(Schedulers.computation())
      .doOnNext(new Action1<String>() {
        @Override
        public void call(String t1) {
          System.out.println("successfully executed");
        }
      });
}

代码示例来源：origin: com.microsoft.azure/azure-mgmt-resources

/**
 * Gets the specified resource provider.
 *
 * @param resourceProviderNamespace The namespace of the resource provider.
 * @throws IllegalArgumentException thrown if parameters fail the validation
 * @throws CloudException thrown if the request is rejected by server
 * @throws RuntimeException all other wrapped checked exceptions if the request fails to be sent
 * @return the ProviderInner object if successful.
 */
public ProviderInner get(String resourceProviderNamespace) {
  return getWithServiceResponseAsync(resourceProviderNamespace).toBlocking().single().body();
}