Iterator 与 fail-fast：集合遍历的安全边界

2024-03-15 02:17:33 ERROR OrderSyncService - sync failed
java.util.ConcurrentModificationException
    at java.util.ArrayList$Itr.checkForComodification(ArrayList.java:1013)
    at java.util.ArrayList$Itr.next(ArrayList.java:965)
    at com.biz.service.OrderSyncService.processOrders(OrderSyncService.java:87)
    ...

public interface Iterator<E> {

    // 是否还有下一个元素
    boolean hasNext();

    // 返回下一个元素，并将游标后移
    E next();

    // 移除上一次 next() 返回的元素（default 方法抛 UnsupportedOperationException）
    default void remove() {
        throw new UnsupportedOperationException("remove");
    }

    // Java 8 引入：对剩余元素执行 action
    default void forEachRemaining(Consumer<? super E> action) {
        Objects.requireNonNull(action);
        while (hasNext())
            action.accept(next());
    }
}

// 以下两段代码结构完全一致，尽管底层数据结构截然不同

List<String> list = new ArrayList<>(List.of("A", "B", "C"));
Set<String>  set  = new HashSet<>(List.of("A", "B", "C"));

// 遍历 ArrayList —— 底层是数组 + 索引
Iterator<String> it1 = list.iterator();
while (it1.hasNext()) {
    System.out.println(it1.next());
}

// 遍历 HashSet —— 底层是哈希表
Iterator<String> it2 = set.iterator();
while (it2.hasNext()) {
    System.out.println(it2.next());
}

public abstract class AbstractList<E> extends AbstractCollection<E>
    implements List<E> {

    // 结构性修改的计数：add, remove, clear 等操作每次 +1
    // 注意：它是 protected 的，子类可以直接访问
    protected transient int modCount = 0;
}

// add 方法 —— 每次成功添加都让 modCount +1
public boolean add(E e) {
    ensureCapacityInternal(size + 1);  // 内部无条件 modCount++
    elementData[size++] = e;
    return true;
}

private void ensureExplicitCapacity(int minCapacity) {
    modCount++;                    // ★ 无条件递增：每次 add 都 +1
    if (minCapacity - elementData.length > 0)
        grow(minCapacity);         // 真正扩容：newCap = oldCap + (oldCap >> 1)，即 1.5 倍
}

// remove(int index) —— 同样 modCount +1
public E remove(int index) {
    rangeCheck(index);
    modCount++;                    // ← 结构性修改，+1
    E oldValue = elementData(index);
    int numMoved = size - index - 1;
    if (numMoved > 0)
        System.arraycopy(elementData, index + 1, elementData, index, numMoved);
    elementData[--size] = null;  // 尾部置空，帮助 GC
    return oldValue;
}

private class Itr implements Iterator<E> {
    int cursor = 0;                 // 下一个要返回的元素下标
    int lastRet = -1;               // 上一次 next() 返回的元素下标，-1 表示"没有"
    int expectedModCount = modCount; // ★ 创建时对 modCount 拍快照

    public boolean hasNext() {
        return cursor != size;       // 注意：不校验 modCount！
    }

    public E next() {
        checkForComodification();       // ★ 第一行就校验版本号
        int i = cursor;
        if (i >= size)
            throw new NoSuchElementException();
        Object[] elementData = ArrayList.this.elementData;
        if (i >= elementData.length)  // 数组长度都小于 size 了 → 集合被删过
            throw new ConcurrentModificationException();
        cursor = i + 1;
        return (E) elementData[lastRet = i];
    }

    final void checkForComodification() {
        if (modCount != expectedModCount)
            throw new ConcurrentModificationException();
    }
}

List<String> list = new ArrayList<>();
list.add("A");  // modCount = 1
list.add("B");  // modCount = 2
list.add("C");  // modCount = 3

Iterator<String> it = list.iterator();
// ↑ Itr 构造，expectedModCount = 3（快照）

while (it.hasNext()) {
    String s = it.next();   // 第一次调用 next()：checkForComodification() 通过
    if ("B".equals(s)) {
        list.remove(s);     // ← 用 list 的 remove()！modCount 变成 4
    }
    // 第二次调用 it.next()：
    //   checkForComodification() 发现 modCount(4) != expectedModCount(3)
    //   → 抛出 ConcurrentModificationException!
}

// 源：for (String s : list) { System.out.println(s); }
0:  aload_1
1:  invokeinterface #2  java/util/List.iterator()Ljava/util/Iterator;
6:  astore_2              // 编译器悄悄把迭代器塞进局部变量表
7:  aload_2
8:  invokeinterface #3  java/util/Iterator.hasNext()Z
13: ifeq 36             // hasNext() == false 就跳出循环
16: aload_2
17: invokeinterface #4  java/util/Iterator.next()Ljava/lang/Object;
22: checkcast #5            // 向下转型回 String
25: astore_3                // 存入循环变量 s
26: getstatic #6            // System.out
29: aload_3
30: invokevirtual #7        // println(s)
33: goto 7               // 回到 hasNext() 判断
36: return

public void remove() {
    if (lastRet < 0)                    // ① 还没 next() 就 remove()？
        throw new IllegalStateException();
    checkForComodification();              // ② 先做 fail-fast 校验

    try {
        ArrayList.this.remove(lastRet); // ③ 删"上一次 next() 返回的元素"（内部 modCount++）
        cursor = lastRet;                  // ④ 游标回退到被删位置，避免跳过下一个元素
        lastRet = -1;                   // ⑤ 一次 next 只能配一次 remove
        expectedModCount = modCount;       // ★ ⑥ 同步版本号——"安全"的关键
    } catch (IndexOutOfBoundsException ex) {
        throw new ConcurrentModificationException();
    }
}

List<String> list = new ArrayList<>(List.of("A", "B", "C"));
Iterator<String> it = list.iterator();  // cursor=0, lastRet=-1, expected=3

it.next();  // → "A"，cursor=1, lastRet=0
it.next();  // → "B"，cursor=2, lastRet=1

it.remove(); // 内部：remove(1) → 数组变为 [A, C]，size=2，modCount=4
             //       cursor=1, lastRet=-1, expectedModCount=4（同步！）

it.next();  // → "C"：checkForComodification 4==4 通过；cursor=2
// 结果：A、C 都被遍历到——没有漏元素，也没有抛异常

Iterator<String> it = list.iterator();

it.remove();   // ❌ IllegalStateException：还没 next()，lastRet == -1

// 或：
it.next();     // → "A"
it.remove();   // 删 A，lastRet 置回 -1
it.remove();   // ❌ IllegalStateException：一次 next() 只能配一次 remove()

default boolean removeIf(Predicate<? super E> filter) {
    Objects.requireNonNull(filter);
    boolean removed = false;
    final Iterator<E> each = iterator();   // 拿到自己的迭代器
    while (each.hasNext()) {
        if (filter.test(each.next())) {
            each.remove();          // 用迭代器删 → expectedModCount 自动同步
            removed = true;
        }
    }
    return removed;
}

public boolean removeIf(Predicate<? super E> filter) {
    Objects.requireNonNull(filter);
    final int removeCount;
    final BitSet removeSet;
    final Object[] es = elementData;
    final int size = this.size;
    final int expectedModCount = modCount;
    removeSet = new BitSet(size);          // ① BitSet 记录"要删的下标"

    // ② 一趟扫描：循环条件里就盯着 modCount，被并发改了就提前退出
    for (int i = 0; modCount == expectedModCount && i < size; i++) {
        final E element = (E) es[i];
        if (filter.test(element)) {
            removeSet.set(i);
        }
    }
    if (modCount != expectedModCount)       // ③ 扫描期被并发修改 → 抛 CME（列表还没动）
        throw new ConcurrentModificationException();

    removeCount = removeSet.cardinality();
    if (removeCount > 0) {
        final int newSize = size - removeCount;
        // ④ 双指针左移压缩：把"不删的"依次搬到前面
        for (int i = 0, j = 0; i < size && j < newSize; i++) {
            if (!removeSet.get(i)) {
                es[j++] = es[i];
            }
        }
        // ⑤ 尾部置空，交给 GC
        for (int i = newSize; i < size; i++) {
            es[i] = null;
        }
        this.size = newSize;
        if (modCount != expectedModCount)     // ⑥ 压缩前再校验一次
            throw new ConcurrentModificationException();
        modCount++;                              // ⑦ 整个操作只 +1 一次！
    }
    return removeCount > 0;
}

Iterator<String> it = list.iterator();
list.removeIf(s -> s.startsWith("tmp"));  // ✅ 自身安全，但 modCount +1 了
it.next();   // ❌ CME！你手上这个迭代器的 expectedModCount 已过期

// 结论：removeIf 只保证"自己这一趟"不出事，
//      任何其他正在进行的遍历，都会被它的一次 modCount++ 打断

public class CopyOnWriteArrayList<E> implements List<E>, RandomAccess {

    // 底层数组：volatile 保证跨线程可见性；没有任何 modCount
    private transient volatile Object[] array;

    // add：加锁 → 复制整数组 → 追加 → 原子替换引用
    public boolean add(E e) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            Object[] elements = getArray();
            int len = elements.length;
            Object[] newElements = Arrays.copyOf(elements, len + 1); // ← 全量复制
            newElements[len] = e;
            setArray(newElements);     // volatile 写，原子替换引用
            return true;
        } finally {
            lock.unlock();
        }
    }

    // iterator：直接把"当前数组引用"交给迭代器——这就是快照
    public Iterator<E> iterator() {
        return new COWIterator<>(getArray(), 0);
    }
}

static final class COWIterator<E> implements ListIterator<E> {
    private final Object[] snapshot;   // ★ 快照：创建时固定，之后不可变
    private int cursor;

    public E next() {
        if (!hasNext()) throw new NoSuchElementException();
        return (E) snapshot[cursor++];
    }

    public void remove() {
        throw new UnsupportedOperationException();  // ★ 快照是只读的！
    }
}

// 想删？通过集合自身删（会再复制一次），正在进行的遍历不受影响：
cowList.remove("B");   // 旧快照里 B 还在——遍历结果可能"过期"

// ─── CopyOnWriteArrayList：严格快照，看不到任何后续修改 ───
CopyOnWriteArrayList<String> cowList = new CopyOnWriteArrayList<>(List.of("A", "B", "C"));
Iterator<String> cowIt = cowList.iterator();
cowList.add("D");                  // 触发复制，旧数组快照不变
while (cowIt.hasNext()) System.out.print(cowIt.next() + " ");
// 输出：A B C（永远看不到 D）

// ─── ConcurrentHashMap：弱一致，可能看到也可能看不到 ───
ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();
map.put("A", 1); map.put("B", 2); map.put("C", 3);
Iterator<String> chmIt = map.keySet().iterator();
map.put("D", 4);                  // 修改底层 table
while (chmIt.hasNext()) System.out.print(chmIt.next() + " ");
// 输出：A B C D 或 A B C（取决于遍历到那个桶时的时序）

// 场景 A：foreach 里删"当前元素后面还有元素" → 必抛
List<String> list = new ArrayList<>(List.of("A", "B", "C", "D"));
for (String s : list) {
    if ("B".equals(s)) list.remove(s);  // 删 B 后还有 C、D → 下次 next() 抛 CME
}

// 场景 B：删除后 cursor 恰好等于新 size → 可能不抛！
List<String> list2 = new ArrayList<>(List.of("A", "B", "C"));
Iterator<String> it = list2.iterator();
it.next();  // A，cursor=1
it.next();  // B，cursor=2
list2.remove("C");  // 删的是游标之后的尾部元素 → size=2
// 下一轮 hasNext()：cursor(2) != size(2) 为 false → 循环结束，无 CME
// 但这纯属侥幸：CME 的触发点只在 next()/remove() 内部，hasNext() 从不检查

List<String> list = new ArrayList<>(List.of("A", "B", "C"));
for (String s : list) {
    if ("B".equals(s)) {
        list.remove(s);   // CME！一个线程也会抛
    }
}
// 面试结论：fail-fast 不是线程安全机制，它是检测"错误用法"的手段
// 单线程的错误用法同样触发 CME——"并发"两个字只是名字，不代表只在并发下发生

List<String> parent = new ArrayList<>(List.of("A", "B", "C", "D"));
List<String> sub = parent.subList(1, 3);  // [B, C] —— 视图，不是副本！

parent.add("E");       // 父集合 modCount +1
System.out.println(sub.get(0));
// → ConcurrentModificationException！
// SubList 每次操作前都检查"创建时的父 modCount"是否与当前一致

Map<String, Integer> map = new HashMap<>();
map.put("a", 1); map.put("b", 2);

for (Map.Entry<String, Integer> e : map.entrySet()) {
    map.put("c", 3);  // ❌ CME！新增 key 是结构性修改（modCount++）
}
// 注意：put 已有 key（覆盖 value）不是结构性修改，不会抛——又是"尽力而为"的体现

// ✅ 正确：
//  1) entrySet().removeIf(...)  只删不增
//  2) ConcurrentHashMap        遍历中随便 put
//  3) 先收集到新集合，遍历完再 putAll 合并

// ① 单线程批量条件删除：removeIf 一趟 O(n)，首选
orders.removeIf(o -> o.getStatus() == Status.CANCELLED);

// ② 需要边删边做业务（记日志 / 发消息）：显式 Iterator
Iterator<Order> it = orders.iterator();
while (it.hasNext()) {
    Order o = it.next();
    if (o.isExpired()) {
        auditLog.info("expire order {}", o.getId());
        it.remove();
    }
}

// ③ 读多写极少：监听器 / 订阅者列表 → CopyOnWriteArrayList
private final CopyOnWriteArrayList<OrderListener> listeners =
    new CopyOnWriteArrayList<>();
void notifyListeners(Order o) {
    for (OrderListener l : listeners) l.onOrder(o);  // 遍历中注册/注销都安全
}

// ④ 高并发 KV：ConcurrentHashMap（不要 HashMap + synchronized 自欺欺人）
ConcurrentHashMap<String, Session> sessions = new ConcurrentHashMap<>();
sessions.computeIfAbsent(token, Session::create);   // 原子读写

// ⑤ 必须遍历共享 ArrayList 快照：先拷贝，别在锁里遍历大集合
synchronized (orderLock) {
    for (Order o : new ArrayList<>(orders)) { // 快照遍历，锁内只做拷贝
        // 处理期间 orders 随便被改，互不影响
    }
}

// ❌ 并行流中修改源集合——结果不可预测
list.parallelStream()
    .filter(s -> {
        list.add("X");   // 并发修改源 list！可能 CME、可能丢数据、可能"看起来正常"
        return true;
    })
    .count();

// ✅ Stream 必须无副作用：用中间流/新集合，最后一次性合并
List<String> result = list.stream()
    .filter(s -> s.length() > 1)
    .collect(Collectors.toList());