gistfile1.java

public class Test {
  public static class ArrayListWrapper {
    int id;
    ArrayList<Integer> list = new ArrayList<>();

    public ArrayListWrapper(int id) {
      this.id = id;
    }

    public void add(int i) {
      list.add(i);
    }

    @Override
    public String toString() {
      return list.toString();
    }

    @Override
    public boolean equals(Object o) {
      if (this == o) return true;
      if (o == null || getClass() != o.getClass()) return false;

      ArrayListWrapper li = (ArrayListWrapper) o;

      return id == li.id;

    }

    @Override
    public int hashCode() {
      return id;
    }
  }

  static int num = 100_000;
  static Observable<Integer>[] obs = new Observable[]{
    Observable.range(0, num),
    Observable.range(0, num),
    Observable.range(0, num),
    Observable.range(0, num),
    Observable.range(0, num)
  };

  public static void main(String[] args) {
    for (int i = 0; i < 100; i++) {
      long now = System.currentTimeMillis();
      plain();
      System.out.println("plain " + (System.currentTimeMillis() - now));
      now = System.currentTimeMillis();
      rx();
      System.out.println("rx " + (System.currentTimeMillis() - now));
    }
  }

  public static void plain() {
    HashMap<Integer, ArrayListWrapper> map = new HashMap<>();
    for (Observable<Integer> o : obs) {
      for (int number : o.toBlocking().toIterable()) {
        ArrayListWrapper list = map.get(number);
        if (list == null) {
          list = new ArrayListWrapper(number);
          map.put(number, list);
        }
        list.add(number);
      }
    }
    AtomicInteger count = new AtomicInteger();
    map.entrySet().stream().forEach(e -> count.incrementAndGet());
  }


  public static void rx() {
    AtomicInteger count = new AtomicInteger();
    Observable<ArrayListWrapper> result = Observable.from(new Integer[]{0, 1, 2})
      .flatMap(i -> obs[i])
      .groupBy(i -> new ArrayListWrapper(i))
      .flatMap(group ->
        group.collect(() -> group.getKey(), (list, number) -> list.add(number)));
    result.toBlocking().forEach(l -> count.incrementAndGet());

//    Observable<Map<Integer, Collection<Integer>>> result = Observable.from(new Integer[]{0, 1, 2})
//      .flatMap(i -> obs[i])
//      .toMultimap(integer -> integer);
//    result.toBlocking().forEach(m -> m.entrySet().stream().forEach(System.out::println));
  }
}