利用RxJava避免可变状态

Question

我正在学习RxJava，但我遇到了避免可变状态的问题。

我正在解决的问题很简单：有一个条目的输入流和一个条目组的输入流。每个项目都属于一个组（有一个组标识符），并且有一些数据。每个组都有一个标识符和一些数据。许多项目可能属于同一组。目标是将这些输入流组合成一个（项、组）对的输出流，以便：

• 只有当项目及其组都已知时，才会发出（项目，组）对
• 每次接收更新的项目数据时，必须发出更新的（项目、组）对
• 当接收到该组的更新数据时，必须发出与属于该组的所有项相对应的（项，组）对

下面是一个工作实现（ItemWithGroup是一个表示（项、组）对的类）：

public class StateMutationWithinOperator {
    private Map<Integer, Group> allGroups = new HashMap<>();
    private Map<Integer, List<Item>> allItems = new HashMap<>();

    public Observable<ItemWithGroup> observe(Observable<Item> items, Observable<Group> groups) {
        return Observable.merge(
            items.flatMap(this::processItem), 
            groups.flatMap(this::processGroup));
    }

    private Observable<ItemWithGroup> processItem(Item item) {
        allItems.computeIfAbsent(item.groupId, missing -> new ArrayList<>())
                .add(item);

        return Observable.just(allGroups)
                .filter(groups -> groups.containsKey(item.groupId))
                .map(groups -> new ItemWithGroup(item, groups.get(item.groupId)));
    }

    private Observable<ItemWithGroup> processGroup(Group group) {
        allGroups.put(group.id, group);

        return Observable.just(allItems)
                .filter(items -> items.containsKey(group.id))
                .flatMapIterable(items -> items.get(group.id))
                .map(item -> new ItemWithGroup(item, group));
    }
}

通过避免更改存储在allGroups和allItems字段中的共享状态，我希望避免在processItem（）和processGroup（）中产生副作用。我如何实现这一点？

提前道谢！

Answer 1

以下是根据@phoenixwang的建议进行的改进：

public class StateMutationWithinOperator {
  private Map<Integer, Group> allGroups = new HashMap<>();
  private Map<Integer, List<Item>> allItems = new HashMap<>();

  public Observable<ItemWithGroup> observe(Observable<Item> inputItems,
                                           Observable<Group> inputGroups) {
    Observable<ItemWithGroup> processedItems = inputGroups.publish(groups -> {
      groups.subscribe(group -> allGroups.put(group.id, group));

      return groups.flatMap(this::processGroup);
    });

    Observable<ItemWithGroup> processedGroups = inputItems.publish(items -> {
      items.subscribe(item
          -> allItems
          .computeIfAbsent(item.groupId, missing -> new ArrayList<>())
          .add(item));

      return items.flatMap(this::processItem);
    });

    return Observable.merge(processedItems, processedGroups);
  }

  private Observable<ItemWithGroup> processItem(Item item) {
    return Observable.just(allGroups)
        .filter(groups -> groups.containsKey(item.groupId))
        .map(groups -> new ItemWithGroup(item, groups.get(item.groupId)));
  }

  private Observable<ItemWithGroup> processGroup(Group group) {
    return Observable.just(allItems)
        .filter(items -> items.containsKey(group.id))
        .flatMapIterable(items -> items.get(group.id))
        .map(item -> new ItemWithGroup(item, group));
  }
}

它更好，但是publish（）运算符仍然会使共享状态发生变化。我可以看到这样做的一个问题是，我们可能同时从两个不同的线程接收一个项目和一个组，并最终同时读写到同一个映射。