JavaRX 分页 - 在每次交互中而不是在最后观察 - 通用分页器

Question

我正在使用分页 API。我使用了 Adam Millerchip 提供的以下解决方案而且效果很好。

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Optional;
import java.util.function.Function;

import io.reactivex.Flowable;
import io.reactivex.Single;
import io.reactivex.processors.BehaviorProcessor;

public class Pagination {

    // Fetch all pages and return the items contained in those pages, using the provided page fetcher function
    public static <T> Flowable<T> fetchItems(Function<Integer, Single<Page<T>>> fetchPage) {
        // Processor issues page indices
        BehaviorProcessor<Integer> processor = BehaviorProcessor.createDefault(0);
        // When an index number is issued, fetch the corresponding page
        return processor.concatMap(index -> fetchPage.apply(index).toFlowable())
                        // when returning the page, update the processor to get the next page (or stop)
                        .doOnNext(page -> {
                            if (page.hasNext()) {
                                processor.onNext(page.getNextPageIndex());
                            } else {
                                processor.onComplete();
                            }
                        })
                        .concatMapIterable(Page::getElements);
    }

    public static void main(String[] args) {
        fetchItems(Pagination::examplePageFetcher).subscribe(System.out::println);
    }

    // A function to fetch a page of our paged data
    private static Single<Page<String>> examplePageFetcher(int index) {
        return Single.just(pages.get(index));
    }

    // Create some paged data
    private static ArrayList<Page<String>> pages = new ArrayList<>(3);

    static {
        pages.add(new Page<>(Arrays.asList("one", "two"), Optional.of(1)));
        pages.add(new Page<>(Arrays.asList("three", "four"), Optional.of(2)));
        pages.add(new Page<>(Arrays.asList("five"), Optional.empty()));
    }

    static class Page<T> {
        private List<T> elements;
        private Optional<Integer> nextPageIndex;

        public Page(List<T> elements, Optional<Integer> nextPageIndex) {
            this.elements = elements;
            this.nextPageIndex = nextPageIndex;
        }

        public List<T> getElements() {
            return elements;
        }

        public int getNextPageIndex() {
            return nextPageIndex.get();
        }

        public boolean hasNext() {
            return nextPageIndex.isPresent();
        }
    }
}

但我有两个问题:

• 在此实现中，当加载所有页面时，元素将在最后 (subscribe(System.out::println)) 进行处理。如果收集的数据很多，这可能会导致内存问题。我更愿意在加载它们时立即处理它们(数据库保存)(在 .doOnNext(page -> { }) 中)。我已经能够做到这一点，但以“肮脏的方式”(在doOnNext)。我该怎么做？

• 在“page”类的实现中，我使用自定义的 Gson 反序列化器。而且我不知道如何处理通用数据。我不得不写“list.add((MyGenericClass)context.deserialize(anArray.getAsJsonObject(), MyGenericClass.class));”我想要类似“list.add((T)context.deserialize(anArray.getAsJsonObject(), T.class));”的内容。我怎样才能让事情真正通用？

  public static JsonDeserializer<Paginator> deserializer = new JsonDeserializer<Paginator>() {
  @Override
  public Paginator deserialize(JsonElement json, Type typeOfT, JsonDeserializationContext context) throws JsonParseException {
      JsonObject jsonObject = json.getAsJsonObject();
      Paginator paginator = new Paginator(null, Optional.of(1));
      if (jsonObject.get("data") != null && !jsonObject.get("data").isJsonNull()) {
          JsonArray array = jsonObject.getAsJsonArray("data");
          List<MyGenericClass> list = new ArrayList<>();
          for (JsonElement anArray : array) {
              list.add((MyGenericClass)context.deserialize(anArray.getAsJsonObject(), MyGenericClass.class));
          }
          paginator.setData(list);
      }
      paginator.setCurrent_page(jsonAsInt(jsonObject, "current_page",-1));
      paginator.setFrom(jsonAsInt(jsonObject,"from",-1));
      paginator.setTo(jsonAsInt(jsonObject,"to",-1));
      paginator.setTotal(jsonAsInt(jsonObject,"total",-1));
      paginator.setLast_page(jsonAsInt(jsonObject, "last_page", -1));
      paginator.setNextPage(); // calculate next page
      return paginator;
  }
  };
  

Answer 1

回答你的第一个问题:

In this implementation elements are processed at the end (subscribe(System.out::println)) when all pages are loaded."

这是不正确的。响应式(Reactive)编程的全部目的就是避免这种情况。 fetchItems()返回 Flowable<T> ，在有人订阅它之前，它实际上不会获取任何项目。当您订阅某项内容时，每次该项目准备就绪时，订阅者都会收到通知。您应该调用subscribe()并传递一个函数，每次项目准备好时都会调用该函数。在我的示例中，我传递了 System.out::println ，它会打印值，但您可以实现自己的处理程序来保存到数据库。

I would prefer to process them (data base save) immediately when they are loaded (in the .doOnNext(page -> { })

这混淆了发布者和消费者之间的区别。出版商生产元素 - 在我的示例中是 Flowable<T>生产 T 类型的元素。消费者消费发布者生产的元素。 doOnNext() 是发布者的功能。它说“当你发布某些东西时，也会产生这种副作用”。在我的示例中，副作用是发出要获取的下一个页码。您不应该在那里处理数据库保存，您应该编写自己的回调函数( Consumer )或 Subscriber处理它，并将其提供给订阅调用。