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

Observable.blockingForEach介绍

[英]Invokes a method on each item emitted by this Observable and blocks until the Observable completes.

Note: This will block even if the underlying Observable is asynchronous.

This is similar to Observable#subscribe(Observer), but it blocks. Because it blocks it does not need the Observer#onComplete() or Observer#onError(Throwable) methods. If the underlying Observable terminates with an error, rather than calling onError, this method will throw an exception.

The difference between this method and #subscribe(Consumer) is that the onNext action is executed on the emission thread instead of the current thread. Scheduler: blockingForEach does not operate by default on a particular Scheduler. Error handling: If the source signals an error, the operator wraps a checked Exceptioninto RuntimeException and throws that. Otherwise, RuntimeExceptions and Errors are rethrown as they are.
[中]对该可观察对象发出的每个项调用一个方法，并阻塞，直到该可观察对象完成。
*注意：*即使基础的可观察对象是异步的，这也会阻塞。
这类似于Observable#subscribe（观察者），但它会阻止。因为它会阻塞，所以不需要Observer#onComplete（）或Observer#onError（可丢弃）方法。如果基础Observable以错误终止，而不是调用OneError，则此方法将抛出异常。
此方法与#subscribe（Consumer）之间的区别在于，onNext操作是在emission线程而不是当前线程上执行的。Scheduler:blockingForEach默认情况下不会在特定计划程序上运行。错误处理：如果源发出错误信号，操作员将选中的异常包装到RuntimeException中并抛出该异常。否则，运行时异常和错误将按原样重新启动。

代码示例

代码示例来源：origin: ReactiveX/RxJava

private static <K, V> Map<K, Collection<V>> toMap(Observable<GroupedObservable<K, V>> observable) {
  final ConcurrentHashMap<K, Collection<V>> result = new ConcurrentHashMap<K, Collection<V>>();
  observable.blockingForEach(new Consumer<GroupedObservable<K, V>>() {
    @Override
    public void accept(final GroupedObservable<K, V> o) {
      result.put(o.getKey(), new ConcurrentLinkedQueue<V>());
      o.subscribe(new Consumer<V>() {
        @Override
        public void accept(V v) {
          result.get(o.getKey()).add(v);
        }
      });
    }
  });
  return result;
}

代码示例来源：origin: ReactiveX/RxJava

@Test(expected = TestException.class)
public void blockingForEachThrows() {
  Observable.just(1)
  .blockingForEach(new Consumer<Integer>() {
    @Override
    public void accept(Integer e) throws Exception {
      throw new TestException();
    }
  });
}

代码示例来源：origin: ReactiveX/RxJava

private static <T> List<List<T>> toLists(Observable<Observable<T>> observables) {
  final List<List<T>> lists = new ArrayList<List<T>>();
  Observable.concat(observables.map(new Function<Observable<T>, Observable<List<T>>>() {
    @Override
    public Observable<List<T>> apply(Observable<T> xs) {
      return xs.toList().toObservable();
    }
  }))
      .blockingForEach(new Consumer<List<T>>() {
        @Override
        public void accept(List<T> xs) {
          lists.add(xs);
        }
      });
  return lists;
}

代码示例来源：origin: ReactiveX/RxJava

@Override
  public Integer apply(Integer v) throws Exception {
    Observable.just(1).delay(10, TimeUnit.SECONDS).blockingForEach(Functions.emptyConsumer());
    return v;
  }
})

代码示例来源：origin: ReactiveX/RxJava

/**
 * This won't compile if super/extends isn't done correctly on generics.
 */
@Test
public void testCovarianceOfZip() {
  Observable<HorrorMovie> horrors = Observable.just(new HorrorMovie());
  Observable<CoolRating> ratings = Observable.just(new CoolRating());
  Observable.<Movie, CoolRating, Result> zip(horrors, ratings, combine).blockingForEach(action);
  Observable.<Movie, CoolRating, Result> zip(horrors, ratings, combine).blockingForEach(action);
  Observable.<Media, Rating, ExtendedResult> zip(horrors, ratings, combine).blockingForEach(extendedAction);
  Observable.<Media, Rating, Result> zip(horrors, ratings, combine).blockingForEach(action);
  Observable.<Media, Rating, ExtendedResult> zip(horrors, ratings, combine).blockingForEach(action);
  Observable.<Movie, CoolRating, Result> zip(horrors, ratings, combine);
}

代码示例来源：origin: ReactiveX/RxJava

/**
 * This won't compile if super/extends isn't done correctly on generics.
 */
@Test
public void testCovarianceOfCombineLatest() {
  Observable<HorrorMovie> horrors = Observable.just(new HorrorMovie());
  Observable<CoolRating> ratings = Observable.just(new CoolRating());
  Observable.<Movie, CoolRating, Result> combineLatest(horrors, ratings, combine).blockingForEach(action);
  Observable.<Movie, CoolRating, Result> combineLatest(horrors, ratings, combine).blockingForEach(action);
  Observable.<Media, Rating, ExtendedResult> combineLatest(horrors, ratings, combine).blockingForEach(extendedAction);
  Observable.<Media, Rating, Result> combineLatest(horrors, ratings, combine).blockingForEach(action);
  Observable.<Media, Rating, ExtendedResult> combineLatest(horrors, ratings, combine).blockingForEach(action);
  Observable.<Movie, CoolRating, Result> combineLatest(horrors, ratings, combine);
}

代码示例来源：origin: ReactiveX/RxJava

/**
 * Confirm that running on a NewThreadScheduler uses the same thread for the entire stream.
 */
@Test
public void testObserveOnWithNewThreadScheduler() {
  final AtomicInteger count = new AtomicInteger();
  final int _multiple = 99;
  Observable.range(1, 100000).map(new Function<Integer, Integer>() {
    @Override
    public Integer apply(Integer t1) {
      return t1 * _multiple;
    }
  }).observeOn(Schedulers.newThread())
  .blockingForEach(new Consumer<Integer>() {
    @Override
    public void accept(Integer t1) {
      assertEquals(count.incrementAndGet() * _multiple, t1.intValue());
      // FIXME toBlocking methods run on the current thread
      String name = Thread.currentThread().getName();
      assertFalse("Wrong thread name: " + name, name.startsWith("Rx"));
    }
  });
}

代码示例来源：origin: ReactiveX/RxJava

/**
 * Confirm that running on a ThreadPoolScheduler allows multiple threads but is still ordered.
 */
@Test
public void testObserveOnWithThreadPoolScheduler() {
  final AtomicInteger count = new AtomicInteger();
  final int _multiple = 99;
  Observable.range(1, 100000).map(new Function<Integer, Integer>() {
    @Override
    public Integer apply(Integer t1) {
      return t1 * _multiple;
    }
  }).observeOn(Schedulers.computation())
  .blockingForEach(new Consumer<Integer>() {
    @Override
    public void accept(Integer t1) {
      assertEquals(count.incrementAndGet() * _multiple, t1.intValue());
      // FIXME toBlocking methods run on the caller's thread
      String name = Thread.currentThread().getName();
      assertFalse("Wrong thread name: " + name, name.startsWith("Rx"));
    }
  });
}

代码示例来源：origin: ReactiveX/RxJava

.blockingForEach(new Consumer<Integer>() {

代码示例来源：origin: ReactiveX/RxJava

@Test
  public void testWindow() {
    final ArrayList<List<Integer>> lists = new ArrayList<List<Integer>>();

    Observable.concat(
      Observable.just(1, 2, 3, 4, 5, 6)
      .window(3)
      .map(new Function<Observable<Integer>, Observable<List<Integer>>>() {
        @Override
        public Observable<List<Integer>> apply(Observable<Integer> xs) {
          return xs.toList().toObservable();
        }
      })
    )
    .blockingForEach(new Consumer<List<Integer>>() {
      @Override
      public void accept(List<Integer> xs) {
        lists.add(xs);
      }
    });

    assertArrayEquals(lists.get(0).toArray(new Integer[3]), new Integer[] { 1, 2, 3 });
    assertArrayEquals(lists.get(1).toArray(new Integer[3]), new Integer[] { 4, 5, 6 });
    assertEquals(2, lists.size());

  }
}

代码示例来源：origin: ReactiveX/RxJava

@Test(timeout = 2000)
public void testMultiTake() {
  final AtomicInteger count = new AtomicInteger();
  Observable.unsafeCreate(new ObservableSource<Integer>() {
    @Override
    public void subscribe(Observer<? super Integer> observer) {
      Disposable bs = Disposables.empty();
      observer.onSubscribe(bs);
      for (int i = 0; !bs.isDisposed(); i++) {
        System.out.println("Emit: " + i);
        count.incrementAndGet();
        observer.onNext(i);
      }
    }
  }).take(100).take(1).blockingForEach(new Consumer<Integer>() {
    @Override
    public void accept(Integer t1) {
      System.out.println("Receive: " + t1);
    }
  });
  assertEquals(1, count.get());
}

代码示例来源：origin: ReactiveX/RxJava

@Test(timeout = 5000)
public void toObservableNormal() {
  normal.completable.toObservable().blockingForEach(Functions.emptyConsumer());
}

代码示例来源：origin: ReactiveX/RxJava

@Test(timeout = 5000, expected = TestException.class)
public void toObservableError() {
  error.completable.toObservable().blockingForEach(Functions.emptyConsumer());
}

代码示例来源：origin: ReactiveX/RxJava

@Test
  public void testUnsubscribeScan() throws Exception {

    ObservableEventStream.getEventStream("HTTP-ClusterB", 20)
    .scan(new HashMap<String, String>(), new BiFunction<HashMap<String, String>, Event, HashMap<String, String>>() {
      @Override
      public HashMap<String, String> apply(HashMap<String, String> accum, Event perInstanceEvent) {
        accum.put("instance", perInstanceEvent.instanceId);
        return accum;
      }
    })
    .take(10)
    .blockingForEach(new Consumer<HashMap<String, String>>() {
      @Override
      public void accept(HashMap<String, String> pv) {
        System.out.println(pv);
      }
    });

    Thread.sleep(200); // make sure the event streams receive their interrupt
  }
}

代码示例来源：origin: ReactiveX/RxJava

Observable.merge(source).take(6).blockingForEach(new Consumer<Long>() {

代码示例来源：origin: ReactiveX/RxJava

}).blockingForEach(new Consumer<String>() {

代码示例来源：origin: ReactiveX/RxJava

}).blockingForEach(new Consumer<String>() {

代码示例来源：origin: ReactiveX/RxJava

.blockingForEach(new Consumer<Object>() {
  @Override
  public void accept(Object pv) {

代码示例来源：origin: ReactiveX/RxJava

@Test
public void testTakeUnsubscribesOnGroupBy() throws Exception {
  Observable.merge(
    ObservableEventStream.getEventStream("HTTP-ClusterA", 50),
    ObservableEventStream.getEventStream("HTTP-ClusterB", 20)
  )
  // group by type (2 clusters)
  .groupBy(new Function<Event, String>() {
    @Override
    public String apply(Event event) {
      return event.type;
    }
  })
  .take(1)
  .blockingForEach(new Consumer<GroupedObservable<String, Event>>() {
    @Override
    public void accept(GroupedObservable<String, Event> v) {
      System.out.println(v);
      v.take(1).subscribe();  // FIXME groups need consumption to a certain degree to cancel upstream
    }
  });
  System.out.println("**** finished");
  Thread.sleep(200); // make sure the event streams receive their interrupt
}

代码示例来源：origin: ReactiveX/RxJava

.blockingForEach(new Consumer<Object>() {
  @Override
  public void accept(Object pv) {