多轮并行计算怎样拒绝半轮结果：Phaser 与候选版本分开管理

把一轮计算拆给三个工作者，最后一个工作者抛异常，其他两份结果却已经覆盖原数组。即使随后把 Phaser 强制终止，也不能撤回那些写入。屏障控制“什么时候可以继续”，不会替共享结果提供失败回滚。

所以这份多轮计算先分成两套数组：accepted 是已经采用的版本，candidate 是本轮独立候选。所有工作者只读 accepted，各写 candidate 的不重叠范围；协调者确认整轮完成且没有失败，才让候选成为下一轮输入。同步与采用结果是两道边界。

一轮结果怎样交给下一轮

实验按 新值 = 旧值 * 2 + 轮次 更新四个整数，一共三轮。轮次从零开始，第一轮应得到 2、4、6、8，第三轮应得到 12、20、28、36。简单规则让每个结果都可手算，避免让并发机制被业务算法遮住。

协调者一直注册在 Phaser 中，保证工作者还没注册完时阶段不会提前推进。每一轮重新注册有限个任务，任务处理完自己的分片后 arriveAndDeregister；当协调者也 arrive，当前轮的全部注册者才算到齐。下一轮只剩协调者这名常驻参与者，再加入新的有限分片。

这里使用三个执行线程、最多三个同时注册的工作任务，没有把一万个元素各自包装成一个任务。输入最多一万个数、最多十轮、最多八个工作者，避免教程把调度数量与结果保留成本藏到无界列表里。

完整实现与一次失败注入

保存为 RoundCalculation.java，Java 17 下执行 java RoundCalculation.java。第一遍跑完三轮，第二遍固定在轮次一、下标二处抛异常。注入点只服务测试，不依赖随机休眠或碰巧出现的竞态。

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicReference;

public class RoundCalculation {
    static final class BatchFailed extends Exception {
        final int completedRounds;
        final List<Integer> lastAccepted;
        BatchFailed(int completed, int[] accepted, Throwable cause) {
            super("round failed", cause);
            completedRounds = completed;
            lastAccepted = Arrays.stream(accepted).boxed().toList();
        }
    }
    static List<List<Integer>> calculate(int[] input, int workers, int rounds, int failRound)
            throws Exception {
        if (input.length < 1 || input.length > 10000 || workers < 1 || workers > 8
                || rounds < 1 || rounds > 10) throw new IllegalArgumentException("invalid experiment size");
        int[] accepted = input.clone();
        Phaser phaser = new Phaser(1); // Coordinator remains registered across rounds.
        ExecutorService executor = Executors.newFixedThreadPool(workers);
        List<Future<?>> all = new ArrayList<>();
        List<List<Integer>> history = new ArrayList<>();
        AtomicReference<Throwable> failure = new AtomicReference<>();
        try {
            for (int round = 0; round < rounds; round++) {
                final int r = round;
                final int[] source = accepted;
                final int[] candidate = new int[source.length];
                int parties = Math.min(workers, source.length);
                for (int part = 0; part < parties; part++) {
                    final int from = part * source.length / parties;
                    final int to = (part + 1) * source.length / parties;
                    if (phaser.register() < 0) break;
                    try {
                        all.add(executor.submit(() -> {
                            try {
                                for (int i = from; i < to; i++) {
                                    if (Thread.currentThread().isInterrupted()) throw new CancellationException();
                                    if (r == failRound && i == 2) throw new IllegalArgumentException("injected failure");
                                    candidate[i] = Math.addExact(Math.multiplyExact(source[i], 2), r);
                                }
                            } catch (RuntimeException | Error e) {
                                failure.compareAndSet(null, e);
                                phaser.forceTermination();
                                throw e;
                            } finally {
                                if (!phaser.isTerminated()) phaser.arriveAndDeregister();
                            }
                        }));
                    } catch (RejectedExecutionException e) {
                        failure.compareAndSet(null, e);
                        phaser.forceTermination();
                        break;
                    }
                }
                int current = phaser.arrive();
                final int advanced;
                try { advanced = phaser.awaitAdvanceInterruptibly(current, 2, TimeUnit.SECONDS); }
                catch (TimeoutException e) {
                    failure.compareAndSet(null, e);
                    phaser.forceTermination();
                    throw new BatchFailed(history.size(), accepted, failure.get());
                }
                if (advanced < 0 || failure.get() != null)
                    throw new BatchFailed(history.size(), accepted, failure.get());
                accepted = candidate; // Publish only after this entire round passed.
                history.add(Arrays.stream(accepted).boxed().toList());
            }
            for (Future<?> task : all) task.get(2, TimeUnit.SECONDS);
            phaser.arriveAndDeregister();
            return List.copyOf(history);
        } finally {
            phaser.forceTermination();
            all.forEach(task -> task.cancel(true));
            executor.shutdownNow();
            if (!executor.awaitTermination(3, TimeUnit.SECONDS))
                System.err.println("workers still active; cancellation is cooperative");
        }
    }
    public static void main(String[] args) throws Exception {
        List<List<Integer>> success = calculate(new int[]{1,2,3,4}, 3, 3, -1);
        System.out.println("rounds=" + success);
        if (!success.get(2).equals(List.of(12,20,28,36))) throw new AssertionError("round values");
        try {
            calculate(new int[]{1,2,3,4}, 3, 3, 1);
            throw new AssertionError("failure expected");
        } catch (BatchFailed expected) {
            System.out.println("failed completedRounds=" + expected.completedRounds);
            System.out.println("last accepted=" + expected.lastAccepted);
            System.out.println("cause=" + expected.getCause().getClass().getSimpleName());
            if (expected.completedRounds != 1 || !expected.lastAccepted.equals(List.of(2,4,6,8)))
                throw new AssertionError("partial round was accepted");
        }
    }
}

输出为：

rounds=[[2, 4, 6, 8], [5, 9, 13, 17], [12, 20, 28, 36]]
failed completedRounds=1
last accepted=[2, 4, 6, 8]
cause=IllegalArgumentException

失败批次只接受了一轮，最后有效值仍是第一轮结果。第二轮别的分片可能已经算出若干 candidate 值，但它们没有进入 lastAccepted。测试没有要求其他分片完全没动过，而是断言半轮结果始终没有被采用，这才是这项计算合同需要的性质。

成功路径在最后实际读取全部 Future，任务的异常不会因为只观察了屏障就被丢掉。正常采用版本已经由阶段完成与共享失败记录控制；Future 读取则进一步完成任务执行状态的收尾。两条证据分别回答“本轮数据可采用”与“提交的任务都已经结束”。

各自写数组片段，为什么不用给每个元素加锁

from 和 to 采用半开范围划分，不同任务不会写同一个下标；source 在这一轮里不再改变。阶段推进之前的工作者动作，与协调者在推进之后的读取具有 Phaser 规定的内存顺序，因此协调者不能绕过屏障提前读取 candidate。

没有每元素锁，不等于数组可以任意共享写。若把 accepted 指向同一个原地更新数组，后面的分片可能读到前面刚写的新值，算法就不再是从同一旧版本计算。若分片边界重叠，即使每个赋值本身原子，也不能证明最终值属于所需轮次。

输入在进入计算时 clone，防止外部调用方随后改原数组影响第一轮；历史结果转换成不可修改的整数列表，也不把内部候选数组交给调用方。这里的值都是 int，因而不存在列表元素继续改变对象状态的问题。扩展成可变业务对象时，要重新确定值投影或复制合同。

数学运算采用 addExact 和 multiplyExact。算术溢出会作为明确失败进入相同的整轮拒绝流程，而不是悄悄绕回一个错误整数。并发设计只保护处理顺序，不会让溢出的数值变正确；候选数组正好使这类普通业务失败也不会污染已采用版本。

失败者不能只注销然后消失

如果工作者失败后仅在 finally 注销，其他参与者仍可能正常到齐，协调者把没有填完的 candidate 当成成功。程序先保存失败原因，再 forceTermination；协调者看到负阶段值或失败记录，就拒绝本轮。这份检查不能只靠“任务数量最后变少了”来推断。

AtomicReference 保存第一个观测到的失败，目的是给这次拒绝提供原因，不是证明它一定是时间上最早发生的根因。多个任务几乎同时失败时，第一条成功写入的异常只是代表证据；需要完整诊断时可以收集有限数量的错误，但不应让日志无界增长。

强制终止不代表中断工作线程，所以 finally 仍取消所有 Future 并关闭执行器。每个分片还在循环里响应中断。真实计算若进入一个无法中断的外部调用，必须给那项操作自己的截止时间；同步器终止只能停止后续阶段等待，不能保证外部动作已被撤销。

任务提交也可能失败。代码在注册之后若遇到拒绝，直接记录失败并终止当前 Phaser，不再要求它慢慢等一份永远不会启动的注册资格。在一个被其他组件关闭的共享执行器上，这条路径尤其值得验证；本例执行器由方法拥有，减少了生命周期交错。

等待预算覆盖哪里，需要明说

每轮屏障等待最多两秒，这不是三轮合计两秒。分片算法很短，预算用于避免演示挂住；如果产品要求整批总时间上限，应在方法开头建立一个总 deadline，每次等待传入剩余时间。不同预算模型会影响调用方重试与用户提示，不能只把变量命名为 timeout 就结束设计。

超时发生时，accepted 仍指向上一份有效版本，candidate 被丢弃；调用线程被中断也会进入 finally 结束本批。这里没有提供自动从某一轮恢复的接口，重新运行会从明确输入开始。若要断点续算，需要保存输入版本、轮次和算法参数，而不是只保存一个 Phaser 的阶段号。

Phaser 的阶段计数是进程内同步状态，不适合作为持久业务批次标识。终止后同步方法返回负数，也不能把这份对象重用给下一次独立批次。本例每次 calculate 都创建自己的同步器，失败不会把后一次调用永久绑在旧对象上。

这份局部采用协议没有承诺外部事务

历史列表只是内存结果。若工作者在计算中直接写数据库、发送消息或覆盖文件，candidate 数组隔离仍然撤不回外部副作用。应让工作者输出受控候选，由明确的提交阶段处理外部写入，并按存储系统能力设计原子性、幂等或补偿。

如果每一轮都需要对外发布，后续失败可能留下已经发布的前几轮，这应成为可见业务状态。不能因为函数最后抛异常，就向用户声称整批什么都没发生。教程用 completedRounds 与 lastAccepted 明确暴露局部进展，接入真正业务时也应保留可解释的版本边界。

对于只运行一轮的独立任务集合，完成服务或 Future 汇总通常更直接；本例保留 Phaser，是因为协调者要跨轮建立新的参与者并利用阶段推进。选择它的理由应来自多轮依赖，而不是想把所有并发工具都用一遍。最后那条失败断言比成功输出更值得长期保存，它能及时发现维护时把候选数组改回原地更新的退步。