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

Flowable.collectInto介绍

[英]Collects items emitted by the finite source Publisher into a single mutable data structure and returns a Single that emits this structure.

This is a simplified version of reduce that does not need to return the state on each pass.

Note that this operator requires the upstream to signal onComplete for the accumulator object to be emitted. Sources that are infinite and never complete will never emit anything through this operator and an infinite source may lead to a fatal OutOfMemoryError. Backpressure: This operator does not support backpressure because by intent it will receive all values and reduce them to a single onNext. Scheduler: collectInto does not operate by default on a particular Scheduler.
[中]将有限源发布服务器发出的项收集到单个可变数据结构中，并返回发出此结构的单个数据结构。
这是reduce的简化版本，不需要在每次传递时返回状态。
请注意，此运算符要求上游发出信号onComplete，以发出累加器对象。无限且永远不完整的源永远不会通过此运算符发出任何信息，无限源可能导致致命的OutOfMemoryError。背压：此运算符不支持背压，因为它将接收所有值，并将它们减少到单个onNext。调度程序：默认情况下，collectInto不会在特定调度程序上运行。

代码示例

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

@Test(expected = NullPointerException.class)
public void collectIntoInitialNull() {
  just1.collectInto(null, new BiConsumer<Object, Integer>() {
    @Override
    public void accept(Object a, Integer b) { }
  });
}

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

@Test(expected = NullPointerException.class)
public void collectIntoCollectorNull() {
  just1.collectInto(1, null);
}

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

@SuppressWarnings("unchecked")
@Test
public void collectInto() {
  Flowable.just(1, 1, 1, 1, 2)
  .collectInto(new HashSet<Integer>(), new BiConsumer<HashSet<Integer>, Integer>() {
    @Override
    public void accept(HashSet<Integer> s, Integer v) throws Exception {
      s.add(v);
    }
  })
  .test()
  .assertResult(new HashSet<Integer>(Arrays.asList(1, 2)));
}

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

@SuppressWarnings("unchecked")
@Test
public void collectIntoFlowable() {
  Flowable.just(1, 1, 1, 1, 2)
  .collectInto(new HashSet<Integer>(), new BiConsumer<HashSet<Integer>, Integer>() {
    @Override
    public void accept(HashSet<Integer> s, Integer v) throws Exception {
      s.add(v);
    }
  })
  .toFlowable()
  .test()
  .assertResult(new HashSet<Integer>(Arrays.asList(1, 2)));
}

代码示例来源：origin: com.microsoft.rest.v2/client-runtime

/**
 * Collects byte buffers emitted by a Flowable into a byte array.
 * @param content A stream which emits byte buffers.
 * @return A Single which emits the concatenation of all the byte buffers given by the source Flowable.
 */
public static Single<byte[]> collectBytesInArray(Flowable<ByteBuffer> content) {
  return content.collectInto(Unpooled.buffer(), (buf, bb) -> buf.writeBytes(bb.slice())).map(out -> {
    try {
      if (out.array().length == out.readableBytes()) {
        return out.array();
      } else {
        byte[] arr = new byte[out.readableBytes()];
        out.readBytes(arr);
        return arr;
      }
    } finally {
      out.release();
    }
  });
}

代码示例来源：origin: com.microsoft.rest.v2/client-runtime

private Single<ByteBuf> collectContent() {
  ByteBuf allContent = null;
  String contentLengthString = headerValue("Content-Length");
  if (contentLengthString != null) {
    try {
      int contentLength = Integer.parseInt(contentLengthString);
      allContent = Unpooled.buffer(contentLength);
    } catch (NumberFormatException ignored) {
    }
  }
  if (allContent == null) {
    allContent = Unpooled.buffer();
  }
  return contentStream.collectInto(allContent, new BiConsumer<ByteBuf, ByteBuf>() {
    @Override
    public void accept(ByteBuf allContent, ByteBuf chunk) throws Exception {
      //use try-finally to ensure chunk gets released
      try {
        allContent.writeBytes(chunk);
      }
      finally {
        chunk.release();
      }
    }
  });
}

代码示例来源：origin: io.micronaut/management

/**
   * @param definitions The bean definitions
   * @return A {@link Single} that wraps a Map
   */
  protected Single<Map<String, Object>> getBeans(Collection<BeanDefinition<?>> definitions) {
    Map<String, Object> beans = new ConcurrentHashMap<>(definitions.size());

    return Flowable
      .fromIterable(definitions)
      .subscribeOn(Schedulers.from(executorService))
      .collectInto(beans, (map, definition) ->
        map.put(definition.getClass().getName(), beanDefinitionData.getData(definition))
      );
  }
}

代码示例来源：origin: io.micronaut/micronaut-management

/**
   * @param definitions The bean definitions
   * @return A {@link Single} that wraps a Map
   */
  protected Single<Map<String, Object>> getBeans(Collection<BeanDefinition<?>> definitions) {
    Map<String, Object> beans = new ConcurrentHashMap<>(definitions.size());

    return Flowable
      .fromIterable(definitions)
      .subscribeOn(Schedulers.from(executorService))
      .collectInto(beans, (map, definition) ->
        map.put(definition.getClass().getName(), beanDefinitionData.getData(definition))
      );
  }
}

代码示例来源：origin: io.micronaut/micronaut-management

@Override
public Publisher<Map<String, Object>> getData(Stream<UriRoute> routes) {
  List<UriRoute> routeList = routes.collect(Collectors.toList());
  Map<String, Object> routeMap = new ConcurrentHashMap<>(routeList.size());
  return Flowable
    .fromIterable(routeList)
    .subscribeOn(Schedulers.from(executorService))
    .collectInto(routeMap, (map, route) ->
      map.put(getRouteKey(route), routeData.getData(route))
    ).toFlowable();
}

代码示例来源：origin: io.micronaut/management

@Override
public Publisher<Map<String, Object>> getData(Stream<UriRoute> routes) {
  List<UriRoute> routeList = routes.collect(Collectors.toList());
  Map<String, Object> routeMap = new ConcurrentHashMap<>(routeList.size());
  return Flowable
    .fromIterable(routeList)
    .subscribeOn(Schedulers.from(executorService))
    .collectInto(routeMap, (map, route) ->
      map.put(getRouteKey(route), routeData.getData(route))
    ).toFlowable();
}

代码示例来源：origin: org.infinispan/infinispan-core

public static <K, V> Set<MarshalledEntry<K, V>> allEntries(AdvancedLoadWriteStore<K, V> cl, Predicate<K> filter) {
 return Flowable.fromPublisher(cl.publishEntries(filter, true, true))
    .collectInto(new HashSet<MarshalledEntry<K, V>>(), Set::add)
    .blockingGet();
}