Comment puis-je créer un séparateur de pagination à usage général?

La raison principale, votre spliterator ne vous rapproche pas de votre objectif, est qu'il essaie de diviser les pages, plutôt que l'espace de l'élément source. Si vous connaissez le nombre total d'éléments et avez une source permettant d'extraire une page via offset et limit, la forme la plus naturelle de spliterator est d'encapsuler une plage dans ces éléments, par exemple via offset et limit ou end. Ensuite, le Fractionnement signifie simplement diviser cette plage , en adaptant le décalage de votre spliterator à la division position et création d'un nouveau spliterator représentant le préfixe, de "Old offset" à la position de division.

Before splitting:
      this spliterator: offset=x, end=y
After splitting:
      this spliterator: offset=z, end=y
  returned spliterator: offset=x, end=z

x <= z <= y

Tandis que dans le meilleur des cas, z est exactement au milieu entre les x et y, l'équilibre se divise, mais dans notre cas, nous allons légèrement adapter à produire des ensembles multiples de la taille de la page.

Cette logique fonctionne sans avoir besoin de récupérer des pages, donc si vous reportez la récupération des pages jusqu'au moment, le framework veut commencer la traversée, c'est-à-dire après, le fractionnement, les opérations d'extraction peuvent s'exécuter en parallèle. Le plus grand obstacle est le fait que vous devez aller chercher la première page afin d'en apprendre davantage sur le nombre total d'éléments. La solution ci-dessous sépare cette première extraction du reste, simplifiant l'implémentation. Bien sûr, il doit transmettre le résultat de cette première récupération de page, qui sera consommée lors de la première traversée (dans le cas séquentiel) ou renvoyée en tant que premier préfixe séparé, en acceptant un scission déséquilibrée à ce stade, mais ne plus avoir à y faire face par la suite.

public class PagingSpliterator<T> implements Spliterator<T> {
    public interface PageFetcher<T> {
        List<T> fetchPage(long offset, long limit, LongConsumer totalSizeSink);
    }
    public static final long DEFAULT_PAGE_SIZE = 100;

    public static <T> Stream<T> paged(PageFetcher<T> pageAccessor) {
        return paged(pageAccessor, DEFAULT_PAGE_SIZE, false);
    }
    public static <T> Stream<T> paged(PageFetcher<T> pageAccessor,
                                      long pageSize, boolean parallel) {
        if(pageSize<=0) throw new IllegalArgumentException();
        return StreamSupport.stream(() -> {
            PagingSpliterator<T> pgSp
                = new PagingSpliterator<>(pageAccessor, 0, 0, pageSize);
            pgSp.danglingFirstPage
                =spliterator(pageAccessor.fetchPage(0, pageSize, l -> pgSp.end=l));
            return pgSp;
        }, CHARACTERISTICS, parallel);
    }
    private static final int CHARACTERISTICS = IMMUTABLE|ORDERED|SIZED|SUBSIZED;

    private final PageFetcher<T> supplier;
    long start, end, pageSize;
    Spliterator<T> currentPage, danglingFirstPage;

    PagingSpliterator(PageFetcher<T> supplier,
            long start, long end, long pageSize) {
        this.supplier = supplier;
        this.start    = start;
        this.end      = end;
        this.pageSize = pageSize;
    }

    public boolean tryAdvance(Consumer<? super T> action) {
        for(;;) {
            if(ensurePage().tryAdvance(action)) return true;
            if(start>=end) return false;
            currentPage=null;
        }
    }
    public void forEachRemaining(Consumer<? super T> action) {
        do {
            ensurePage().forEachRemaining(action);
            currentPage=null;
        } while(start<end);
    }
    public Spliterator<T> trySplit() {
        if(danglingFirstPage!=null) {
            Spliterator<T> fp=danglingFirstPage;
            danglingFirstPage=null;
            start=fp.getExactSizeIfKnown();
            return fp;
        }
        if(currentPage!=null)
            return currentPage.trySplit();
        if(end-start>pageSize) {
            long mid=(start+end)>>>1;
            mid=mid/pageSize*pageSize;
            if(mid==start) mid+=pageSize;
            return new PagingSpliterator<>(supplier, start, start=mid, pageSize);
        }
        return ensurePage().trySplit();
    }
    /**
     * Fetch data immediately before traversing or sub-page splitting.
     */
    private Spliterator<T> ensurePage() {
        if(danglingFirstPage!=null) {
            Spliterator<T> fp=danglingFirstPage;
            danglingFirstPage=null;
            currentPage=fp;
            start=fp.getExactSizeIfKnown();
            return fp;
        }
        Spliterator<T> sp = currentPage;
        if(sp==null) {
            if(start>=end) return Spliterators.emptySpliterator();
            sp = spliterator(supplier.fetchPage(
                                 start, Math.min(end-start, pageSize), l->{}));
            start += sp.getExactSizeIfKnown();
            currentPage=sp;
        }
        return sp;
    }
    /**
     * Ensure that the sub-spliterator provided by the List is compatible with
     * ours, i.e. is {@code SIZED | SUBSIZED}. For standard List implementations,
     * the spliterators are, so the costs of dumping into an intermediate array
     * in the other case is irrelevant.
     */
    private static <E> Spliterator<E> spliterator(List<E> list) {
        Spliterator<E> sp = list.spliterator();
        if((sp.characteristics()&(SIZED|SUBSIZED))!=(SIZED|SUBSIZED))
            sp=Spliterators.spliterator(
                StreamSupport.stream(sp, false).toArray(), IMMUTABLE | ORDERED);
        return sp;
    }
    public long estimateSize() {
        if(currentPage!=null) return currentPage.estimateSize();
        return end-start;
    }
    public int characteristics() {
        return CHARACTERISTICS;
    }
}

Il utilise une interface fonctionnelle PageFetcher spécialisée qui peut être implémentée en appelant la méthode accept du rappel avec la taille totale résultante et en renvoyant une liste d'éléments. Le spliterator de pagination déléguera simplement au spliterator de la liste pour la traversée et dans le cas où la concurrence est significativement plus élevée que le nombre de pages résultant, il peut même bénéficier de la division de ceux-ci séparateurs de page, ce qui implique que les listes d'accès aléatoire, comme ArrayList, sont le type de liste préféré ici.

Adapter votre exemple de code à

private static <T> Stream<T> asSlowPagedSource(long pageSize, List<T> things) {
    return PagingSpliterator.paged( (offset, limit, totalSizeSink) -> {
        totalSizeSink.accept(things.size());
        if(offset>things.size()) return Collections.emptyList();
        int beginIndex = (int)offset;
        assert beginIndex==offset;
        int endIndex = Math.min(beginIndex+(int)limit, things.size());
        System.out.printf("Page %6d-%6d:\t%s%n",
                          beginIndex, endIndex, Thread.currentThread());
        // artificial slowdown
        LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(5));
        return things.subList(beginIndex, endIndex);
    }, pageSize, true);
}

Vous pouvez le tester comme

List<Integer> samples=IntStream.range(0, 555_000).boxed().collect(Collectors.toList());
List<Integer> result =asSlowPagedSource(10_000, samples) .collect(Collectors.toList());
if(!samples.equals(result))
    throw new AssertionError();

Étant donné suffisamment de cœurs de processeur libres, il montrera comment les pages sont récupérées simultanément, donc non ordonnées, alors que le résultat sera correctement dans l'ordre de rencontre. Vous pouvez également tester la concurrence des sous-pages qui s'applique lorsqu'il y a moins de pages:

Set<Thread> threads=ConcurrentHashMap.newKeySet();
List<Integer> samples=IntStream.range(0, 1_000_000).boxed().collect(Collectors.toList());
List<Integer> result=asSlowPagedSource(500_000, samples)
    .peek(x -> threads.add(Thread.currentThread()))
    .collect(Collectors.toList());
if(!samples.equals(result))
    throw new AssertionError();
System.out.println("Concurrency: "+threads.size());