掘金 后端 ( ) • 2024-04-11 15:32
Guava AbstractIterator 业务迭代器玩出新花样：源码解析与进阶实战
AbstractIterator 迭代器 实战 新花样

theme: healer-readable highlight: atom-one-dark

前言

Hi，我是Rike，欢迎来到我的频道👏～

在《万字长文设计模式：迭代器模式 - “__ __启动，带你逛遍每个角落！”》的扩展部分，已简易解释源码且给出实战代码，已经能够满足大部分实战场景。

引用文相关部分也会在本文展示，直阅读本篇即可哟～

但是，近期的新需求使用原先实战代码很不合理，无法满足现有逻辑。我就开始玩起来骚操作了。

本文会讲解源码，并进行实战（通用业务迭代器、RPC调用迭代器）。

同时，给大家提供一个“船新版本”的实现思路，可满足在动态切换数据源的情况下使用（例如上述问题）。

接下来，请欣赏🥳 🧐～

一、Guava AbstractIterator介绍

Google Guava 库，听名字就知道是由Google研发的工具库，旨在简化Java开发中的常见任务了，包含：集合、缓存、原语支持、并发库、通用注释、字符串处理、I/O 等等，总之是一个强大的功能工具库。

Google Guava 库具体功能详见：Github -《Google Guava Wiki》

在此当中，AbstractIterator是Guava提供的一个抽象类，用于简化实现迭代器的过程，使开发者更专注与业务开发。

其主要用途为：

  • 简化迭代器实现：只需关注迭代元素的获取逻辑，无需实现所有的Iterator接口方法。
  • 支持惰性计算：不会一次性加载全部数据，需要时获取本轮数据。
  • 自动处理迭代结束状态：判断迭代过程是否结束，防止重复计算，异常处理。

需注意的是，AbstractIterator类本身不具备线程安全性。在多线程环境下使用时，开发者需要自行添加适当的同步机制或设计模式来确保线程安全。

官方文档如下：Github -《Guava AbstractIterator》

本次代码实现使用Maven版本如下：

  • JDK：11
  • Guava：30.1.1-jre
  • Apache commons-collections4：4.4

二、源码解析

/**
 * This class provides a skeletal implementation of the {@code Iterator} interface, to make this
 * interface easier to implement for certain types of data sources.
 *
 * <p>{@code Iterator} requires its implementations to support querying the end-of-data status
 * without changing the iterator's state, using the {@link #hasNext} method. But many data sources,
 * such as {@link java.io.Reader#read()}, do not expose this information; the only way to discover
 * whether there is any data left is by trying to retrieve it. These types of data sources are
 * ordinarily difficult to write iterators for. But using this class, one must implement only the
 * {@link #computeNext} method, and invoke the {@link #endOfData} method when appropriate.
 *
 * <p>Another example is an iterator that skips over null elements in a backing iterator. This could
 * be implemented as:
 *
 * <pre>{@code
 * public static Iterator<String> skipNulls(final Iterator<String> in) {
 *   return new AbstractIterator<String>() {
 *     protected String computeNext() {
 *       while (in.hasNext()) {
 *         String s = in.next();
 *         if (s != null) {
 *           return s;
 *         }
 *       }
 *       return endOfData();
 *     }
 *   };
 * }
 * }</pre>
 *
 * <p>This class supports iterators that include null elements.
 *
 * @author Kevin Bourrillion
 * @since 2.0
 */
// When making changes to this class, please also update the copy at
// com.google.common.base.AbstractIterator
@GwtCompatible
public abstract class AbstractIterator<T> extends UnmodifiableIterator<T> {
  private State state = State.NOT_READY;

  /** Constructor for use by subclasses. */
  protected AbstractIterator() {}

  private enum State {
    /** We have computed the next element and haven't returned it yet. */
    READY,

    /** We haven't yet computed or have already returned the element. */
    NOT_READY,

    /** We have reached the end of the data and are finished. */
    DONE,

    /** We've suffered an exception and are kaput. */
    FAILED,
  }

  private @Nullable T next;

  /**
   * Returns the next element. <b>Note:</b> the implementation must call {@link #endOfData()} when
   * there are no elements left in the iteration. Failure to do so could result in an infinite loop.
   *
   * <p>The initial invocation of {@link #hasNext()} or {@link #next()} calls this method, as does
   * the first invocation of {@code hasNext} or {@code next} following each successful call to
   * {@code next}. Once the implementation either invokes {@code endOfData} or throws an exception,
   * {@code computeNext} is guaranteed to never be called again.
   *
   * <p>If this method throws an exception, it will propagate outward to the {@code hasNext} or
   * {@code next} invocation that invoked this method. Any further attempts to use the iterator will
   * result in an {@link IllegalStateException}.
   *
   * <p>The implementation of this method may not invoke the {@code hasNext}, {@code next}, or
   * {@link #peek()} methods on this instance; if it does, an {@code IllegalStateException} will
   * result.
   *
   * @return the next element if there was one. If {@code endOfData} was called during execution,
   *     the return value will be ignored.
   * @throws RuntimeException if any unrecoverable error happens. This exception will propagate
   *     outward to the {@code hasNext()}, {@code next()}, or {@code peek()} invocation that invoked
   *     this method. Any further attempts to use the iterator will result in an {@link
   *     IllegalStateException}.
   */
  protected abstract T computeNext();

  /**
   * Implementations of {@link #computeNext} <b>must</b> invoke this method when there are no
   * elements left in the iteration.
   *
   * @return {@code null}; a convenience so your {@code computeNext} implementation can use the
   *     simple statement {@code return endOfData();}
   */
  @CanIgnoreReturnValue
  protected final T endOfData() {
    state = State.DONE;
    return null;
  }

  @CanIgnoreReturnValue // TODO(kak): Should we remove this? Some people are using it to prefetch?
  @Override
  public final boolean hasNext() {
    checkState(state != State.FAILED);
    switch (state) {
      case DONE:
        return false;
      case READY:
        return true;
      default:
    }
    return tryToComputeNext();
  }

  private boolean tryToComputeNext() {
    state = State.FAILED; // temporary pessimism
    next = computeNext();
    if (state != State.DONE) {
      state = State.READY;
      return true;
    }
    return false;
  }

  @CanIgnoreReturnValue // TODO(kak): Should we remove this?
  @Override
  public final T next() {
    if (!hasNext()) {
      throw new NoSuchElementException();
    }
    state = State.NOT_READY;
    T result = next;
    next = null;
    return result;
  }

  /**
   * Returns the next element in the iteration without advancing the iteration, according to the
   * contract of {@link PeekingIterator#peek()}.
   *
   * <p>Implementations of {@code AbstractIterator} that wish to expose this functionality should
   * implement {@code PeekingIterator}.
   */
  public final T peek() {
    if (!hasNext()) {
      throw new NoSuchElementException();
    }
    return next;
  }
}

AbstractIterator内含state、next两个私有成员变量，computeNext()、endOfData()、hasNext()、next()四个核心方法。

成员变量

next变量，存储了下一位迭代元素的数据，在hasNext()判断是否有下一位时获取，使用next()方法时返回元素。

state变量，用于记录迭代器的迭代状态，分为：

  • READY：已计算下一位元素，但仍未返回给调用者。
  • NOT_READY：默认状态，表示尚未计算元素或已返回该元素。即，处于未工作状态。
  • DONE：已到达迭代序列末尾，无法再进行迭代。
  • FAILED：异常状态，迭代器遇到意外且已无法继续。

迭代状态会在核心方法中收到控制，能够更加简洁的控制迭代器的运行，防止意外发生，例如：异常处理后继续运行迭代器、迭代序列结束但仍满足迭代条件继续迭代······。

核心方法

四个核心方法作用如下：

  • hasNext()、next()方法实现了JDK Iterator接口，使用者通过调用二者来判断是否可以获取数据、拿到迭代数据。
  • computeNext()：是暴露在外的接口规范，由开发者自行实现。当没有更多元素或者到达迭代的末尾时，应该调用endOfData()​方法来更新状态。
  • endOfData()：用来标识迭代的结束，即设置迭代状态为DONE​。
computeNext()和endOfData()

computeNext()需要开发者自行实现业务逻辑，在设计时需注意：

  • 判断业务流程中是否有下一位元素。
  • 当无法获取下一元素时，调用endOfData()方法更新迭代状态。

通过文档注释，可知这两个方法是强绑定的。当迭代中没有剩余元素时，实现必须调用endOfData() ，如果不这样做可能会导致无限循环。

hasNext()

在JDK官方Iterator中，会将“判断”和“获取”两个操作分开，分别放进hasNext()、next()中，分开思考设计即可。如上文中所实现的业务迭代器，hasNext()方法，放入纯粹的业务判断逻辑，不进行数据的存储。next()方法，获取前先判断是否还有下一元素，再进行数据获取等一系列操作。

Guava库AbstractIterator类实现的方法却大不相同。

@CanIgnoreReturnValue
@Override
public final boolean hasNext() {
  checkState(state != State.FAILED);
  switch (state) {
    case DONE:
      return false;
    case READY:
      return true;
    default:
  }
  return tryToComputeNext();
}

private boolean tryToComputeNext() {
  state = State.FAILED;
  next = computeNext();
  if (state != State.DONE) {
    state = State.READY;
    return true;
  }
  return false;
}

AbstractIterator将“判断”和“获取”两操作合并到了hasNext()中，通过迭代状态判断是否可以正常取出下一位元素。调用后会先判断迭代器是否为FAIL状态，若是则抛异常禁止继续运行。若为NO_READY，则会通过开发者实现的computeNext()方法获取下一元素，并存储到迭代器中，便于调用者随时取出。

next()
@CanIgnoreReturnValue
@Override
public final T next() {
  if (!hasNext()) {
    throw new NoSuchElementException();
  }
  state = State.NOT_READY;
  T result = next;
  next = null;
  return result;
}

next()方法，会先进行hasNext()的判断，若true取则返回元素并重置迭代器状态。这是防御式编程，为了防止开发者直接调用next()，造成异常状态。

若直接调用两次next()，则会取迭代序列的两位元素，并非一个元素取两次。

因为数据是在hasNext()中获取的，next()只需要取出元素即可。

内存资源管理

在next()方法中有next = null;这句代码，很多文章解释为“帮助 JVM 回收对象”，但并没有进行解释。

对于本句代码，我理解是：为了保证每次迭代完成后，都不会将数据流入下一回合。

  1. 为了节约内存资源的创建。

至于是否帮助JVM回收内存，我认为并不在于这句代码，需要结合next属性、迭代器运行流程、运行时栈内存结构进行分析。代码结构请参考下文的实战代码，序列图如下所示。

图片.png

AbstractPageQueryIterator被实例化后，内存变化会分为两个：在堆内存为该对象分配空间，在pageMethod()的栈内存中创建变量对象（引用堆内存地址）。若成功从computeNext()获取迭代数据时，就会在堆内存中为其分配内存空间，并且将其引用地址赋值给AbstractPageQueryIterator的next属性。

实际上，在本次迭代结束前，迭代元素会一直存储在堆内存中，它会在当前迭代完成后，且无任何对象引用其时回收内存。

因此，next = null;并不会直接对JVM内存回收有影响。

三、基础实现

理解完源码之后，我们开始实战。

官方文档给出的示例如下：

public static Iterator<String> skipNulls(final Iterator<String> in) {
  return new AbstractIterator<String>() {
    protected String computeNext() {
      while (in.hasNext()) {
        String s = in.next();
        if (s != null) {
          return s;
        }
      }
      return endOfData();
    }
  };
}

只需实现computeNext()即可，无需关注其他的。但这其实并不满足我们日常的业务需求。最常见场景就是业务迭代器，例如从数据源中迭代获取数据。

业务迭代器在设计时，我理解需要考虑封装性、通用性、分页逻辑三部分。

  • 封装性：只需关注查询函数，将具体的业务逻辑抽象出来，使得迭代器可以适用于不同的查询场景。
  • 通用性：支持泛型。
  • 分页逻辑：可自定义每次迭代获取数量。

通用业务迭代器

代码如下：

public abstract class AbstractQueryIterator<T> extends AbstractIterator<List<T>> {
    /**
     * 页码
     */
    int pageNo = 1;

    /**
     * 每页展示数量
     */
    int pageSize = 10;

    /**
     * 最大页数
     */
    long maxPageNo = 1000;

    public AbstractQueryIterator() {
        super();
    }

    public AbstractQueryIterator(int pageSize) {
        ProjectAssert.isLessThanNumber(pageSize, 0, CommonEnum.SYSTEM_ERROR);
        this.pageSize = pageSize;
    }

    /**
     * 迭代业务方法
     * @return 数据
     */
    @Override
    protected List<T> computeNext() {
        // 当页数>最大页数时，退出
        if (maxPageNo == 0 || pageNo > maxPageNo) {
            return endOfData();
        }
        // 查询数据
        PageInfo<T> pageInfo = this.query(pageNo++, pageSize);
        if (Objects.isNull(pageInfo) || CollectionUtils.isEmpty(pageInfo.getData())) {
            return endOfData();
        }
        this.setMaxPageNo(pageInfo);
        return pageInfo.getData();
    }

    /**
     * 自定义分类查询方法，只专注与查询数据
     * @param pageNo 页码
     * @param pageSize 每页展示数量
     * @return 数据
     */
    public abstract PageInfo<T> query(int pageNo, int pageSize);

    private void setMaxPageNo(PageInfo<T> pageInfo) {
        if (Objects.nonNull(pageInfo)) {
            long total = pageInfo.getTotal();
            if (total % pageSize == 0) {
                maxPageNo = total / pageSize;
            } else {
                maxPageNo = (total / pageSize) + 1;
            }
        }
    }
}

computeNext()方法的实现，我分为了三部分：

  1. 判断业务是否能取下一元素；
  2. 调用分页查询接口，取出下一位元素；
  3. 更改迭代器属性，并返回元素；

其中分页查询接口仍然采用抽象方法定义，将接口暴露，开发者按需自行定义即可。而业务判断是通过计算是否超过总页数，来判断迭代是否结束。

那么，我们如何使用呢？例如，我们需要条件分页查询。

@Test
public void pageQueryIteratorTest() {
    ModelReqDto reqDto = ModelReqDto.builder()
            .modelField("string")
            .pageNo(2)
            .pageSize(10)
            .build();
    AbstractQueryIterator<ModelDto> nativeIterator = new AbstractQueryIterator<>() {
        // count方式一：初始化时查询一次
        final long total = modelService.countByReqDo(BeanUtil.copy(reqDto, ModelReqDo.class));

        @Override
        public PageInfo<ModelDto> query(int pageNo, int pageSize) {
            ModelReqDo reqDo = BeanUtil.copy(reqDto, ModelReqDo.class);
            reqDo.setPageNo(pageNo);
            reqDo.setPageSize(pageSize);
            List<ModelDto> list = modelService.listByReqDo(reqDo);
            if (CollectionUtils.isEmpty(list)) {
                return PageInfoUtil.emptyPageInfo();
            }
            return PageInfo.<ModelDto>builder()
                    .pageNo(pageNo)
                    .pageSize(pageSize)
                    .total(modelService.countByReqDo(reqDo)) // count方式二：每次查询后都会再次查询个数
                    .data(list)
                    .build();
        }
    };
    while (nativeIterator.hasNext()) {
        List<ModelDto> list = nativeIterator.next();
        // 对list进行业务处理
        System.out.println(list);
    }
}

我们只需要初始化迭代器，在进行迭代获取即可。

在这其中，ModelService的listByReqDo()和countByReqD()方法以及查询实体是示例代码，各位看官可按需自行实现。

RPC查询迭代器

在某些情况下，可能参数是个List存储了多个参数，根据这些参数查询对应数据。上述迭代器的设计思路是对的，但迭代终止时机是不对，同时还存在太多无关逻辑。

以往的实践思路是，对参数进行分割，然后每次对分割后的子参数进行查询调用。如：

Lists.partition(paramList, pageSize).stream()
     .map(param -> {
         // 调用查询，进行其他操作
        return result;
     })
     .collect(Collectors.toList());

根据上边的做法，我们其实可以再迭代器内新增方法引用对象和其入参分割子参数属性。迭代器本身只负责调用方法饮用，不管理方法内部。

因此，代码如下：

/**
 * 用于循环遍历，某个批量获取数据的方法。
 * @author linshangqing
 */
public class RpcQueryIterator<T, R> extends AbstractIterator<List<R>> {

    /**
     * 当前分页索引
     */
    private int indexNo = 0;

    /**
     * 最大分页索引
     */
    private int maxIndexNo = 10;

    /**
     * Rpc方法引用
     */
    private Function<? super List<T>, ? extends List<R>> mapper;

    /**
     * 分割参数List
     */
    private List<List<T>> paramList;


    public RpcQueryIterator(int pageSize, List<T> paramList, Function<? super List<T>, ? extends List<R>> mapper) {
        if (checkParam(paramList, mapper)) {
            paramList = List.of();
            endOfData();
        }
        List<List<T>> partition = Lists.partition(paramList, pageSize); // 注意空指针
        if (CollectionUtils.isNotEmpty(partition)) {
            this.paramList = partition;
            maxIndexNo = partition.size() - 1;
            this.mapper = mapper;
        } else {
            endOfData();
        }
    }

    /**
     * 空迭代器方法
     */
    public static <T, R> RpcQueryIterator<T, R> empty() {
        return new RpcQueryIterator<>(1, Lists.newArrayList(), null) {
            @Override
            protected List<R> computeNext() {
                return endOfData();
            }
        };
    }

    @Override
    protected List<R> computeNext() {
        if (indexNo > maxIndexNo) {
            return endOfData();
        }
        return mapper.apply(paramList.get(indexNo++));
    }

    private boolean checkParam(List<T> paramList, Function<? super List<T>, ? extends List<R>> mapper) {
        return CollectionUtils.isEmpty(paramList) || Objects.isNull(mapper);
    }
}

需注意，因为构造函数无法提前结束，所有某些属性赋值需要考虑空指针异常等情况。

使用示例如下：

public List<ModelRespDto> rpcIteratorList(List<ModelReqDto> reqDtoList) {
    // 1.check
    if (CollectionUtils.isEmpty(reqDtoList)) {
        return Collections.emptyList();
    }
    // 2.init
    List<ModelReqDo> reqDoList = BeanUtil.copyList(reqDtoList, ModelReqDo.class);
    List<ModelRespDto> resultList = Lists.newArrayList();
    // 3.build iterator
    RpcQueryIterator<ModelReqDo, ModelDto> iterator = new RpcQueryIterator<>(ProjectConstant.ITERATOR_DEFAULT_PAGE_SIZE, reqDoList, modelService::listByReqDoList);
    // 4.iterator
    while (iterator.hasNext()) {
        List<ModelDto> next = iterator.next();
        resultList.addAll(BeanUtil.copyList(next, ModelRespDto.class));
    }
    // 5.result
    return resultList;
}

横向对比

两种迭代器，最终的目的都是为了迭代获取数据。但是其设计目的、功能实现却有所不同：

  1. 设计目的：

    • AbstractQueryIterator是实现一个通用的查询迭代器，用于按照指定的页码和每页展示数量执行查询操作。
    • RpcQueryIterator是实现一个用于执行RPC调用的迭代器，每次迭代会根据提供的参数列表进行RPC调用。

  2. 功能实现：

    • AbstractQueryIterator将查询方法抽象化，必须由开发者实现逻辑细节，同时需对分页参数进行处理。
    • RpcQueryIterator并不关心分页相关逻辑，只关心是否可以分批次获取数据。

四、新玩法（解决前言疑问）

新需求大致如下：

  • 迭代获取数据，根据入参选择不同的查询接口（分页条件查询、精确IdList查询....）

上文两种迭代器正好满足当前需求，我的思路如下：

  1. 一开始，我考虑写两套迭代器实现逻辑代码。但后期新增迭代方式，代码就只能兼容堆叠，因此放弃这种实现。
  2. 若使用AbstractQueryIterator，实现两套查询流程。虽然能控制在一套流程内，但会有属性兼容问题（分页属性无法清晰确定）、方法选择问题。虽然可以使用工厂/策略模式解决，但最终代码复杂度过高，后续维护难度大。

最终，我考虑继承通用业务迭代器实现具体问题查询方法，最终控制在一套逻辑。即，利用Java的继承、多态的特性。

目的就在于，特殊问题特殊化，控制规范的同时，让其自由拓展。

代码如下：

public class PreciseIdListQueryIterator<R> extends AbstractQueryIterator<R> {
    /**
     * IdList查询方法
     */
    private final Function<List<Long>, List<R>> function;
    /**
     * 方法入参idList
     */
    private List<List<Long>> idList = new ArrayList<>();

    public PreciseIdListQueryIterator(int pageSize, List<Long> idList, Function<List<Long>, List<R>> function) {
        // 1.check
        if (checkParam(idList, function)) {
            endOfData();
            idList = new ArrayList<>();
        }
        if (pageSize <= 0) {
            pageSize = 10;
        }
        // 2.init
        super.pageNo = 0;
        super.pageSize = pageSize;
        int finalPageSize = pageSize;
        Optional.ofNullable(idList).ifPresent(list -> {
            this.idList = Lists.partition(list, finalPageSize);
            super.maxPageNo = this.idList.size();
        });
        this.function = function;
    }

    /**
     * 校验参数
     *
     * @param idList   idList
     * @param function function
     * @return boolean
     */
    private boolean checkParam(List<Long> idList, Function<List<Long>, List<R>> function) {
        return CollectionUtils.isEmpty(idList) || Objects.isNull(function);
    }

    /**
     * 根据IdList查询数据
     *
     * @param pageNo   页码
     * @param pageSize 每页展示数量
     * @return 分页数据
     */
    @Override
    public PageInfo<R> query(int pageNo, int pageSize) {
        // 1.check
        if (Objects.isNull(function) || pageNo > maxPageNo - 1) {
            endOfData();
            return PageInfoUtil.emptyPageInfo();
        }
        // 2.function apply
        List<R> functionList = function.apply(idList.get(pageNo));
        // 3.result
        return PageInfo.<R>builder()
                .pageNo(pageNo)
                .pageSize(pageSize)
                .total(maxPageNo)
                .data(ListUtils.emptyIfNull(functionList))
                .build();
    }
}

public class PageQueryIterator<T extends PageParam, R> extends AbstractQueryIterator<R> {
    /**
     * 查询参数
     */
    private final T t;

    /**
     * 分页查询方法
     */
    private final Function<T, List<R>> pageQueryFunction;

    public PageQueryIterator(int pageSize, T t, Function<T, Long> countFunction, Function<T, List<R>> pageQueryFunction) {
        // 1.check
        if (this.checkParam(t, countFunction, pageQueryFunction)) {
            endOfData();
        }
        if (pageSize <= 0) {
            pageSize = 10;
        }
        // 2.init
        super.pageSize = pageSize;
        this.pageQueryFunction = pageQueryFunction;
        if (Objects.nonNull(countFunction) && Objects.nonNull(t)) {
            t.setPageSize(pageSize);
            this.t = t;
            super.maxPageNo = countFunction.apply(t);
        } else {
            this.t = null;
            super.maxPageNo = 0;
        }


        Optional.ofNullable(countFunction).ifPresentOrElse(method -> super.maxPageNo = method.apply(t), () -> super.maxPageNo = 0);
    }

    /**
     * 校验参数
     *
     * @param t 查询参数
     * @return boolean
     */
    private boolean checkParam(T t, Function<T, Long> countFunction, Function<T, List<R>> pageQueryFunction) {
        return Objects.isNull(t) || Objects.isNull(countFunction) || Objects.isNull(pageQueryFunction);
    }

    /**
     * 分页条件查询
     *
     * @param pageNo   页码
     * @param pageSize 每页展示数量
     * @return 分页数据
     */
    @Override
    public PageInfo<R> query(int pageNo, int pageSize) {
        // 1.check
        if (Objects.isNull(pageQueryFunction)) {
            endOfData();
            return PageInfoUtil.emptyPageInfo();
        }
        // 2.init
        this.t.setPageNo(pageNo);
        this.t.setPageSize(pageSize);

        // 3.function apply
        List<R> resultList = pageQueryFunction.apply(t);

        // 4.build
        return PageInfo.<R>builder()
                .pageNo(pageNo)
                .pageSize(pageSize)
                .total(maxPageNo)
                .data(ListUtils.emptyIfNull(resultList))
                .build();
    }
}

​​

感谢各位观看，下期见～

参考资料

版权声明：个人学习记录，本博客所有文章均采用 CC-BY-NC-SA 许可协议。转载请注明出处。

若有侵权，请留言联系～

如果您觉得文章对您有帮助，请点击文章正下方的小红心一下。您的鼓励是博主的最大动力！