用 Exchanger 交接双缓冲：空容器、尾批与结束消息

生产者正在填 A 缓冲区，消费者正在处理 B 缓冲区。两边都做完这一轮时，交换引用：消费者拿到装好的 A，生产者拿到已经清空的 B。这个模型只需要两个可复用容器，但前提是双方把交换当作所有权交接，交出去以后不再沿旧引用修改它。

Exchanger 提供的正是两条线程配对交换对象的同步点。它不是队列，不会先存下一批对象等未来某个消费者来取，也不会替业务深拷贝数组。下面从七个整数分成三批开始，把尾批、结束消息、超时和中断走完。

每一次交换，都有两份东西被交出去

初始时生产者拥有空 A，消费者拥有空 B。消费者可以先到交换点等待；生产者把 A 填满后到达，两者互换。返回以后，生产者只能写 B，消费者只能读取 A。消费者处理完 A 后清空它，再把这份空容器带到下一次交换。

因此交换之后，双方收到的不是“同一个对象的两个视图”，而是对方原先提供的对象。只有协议规定两边不再保留可写别名，才能把“此时谁能修改”说清。API 的同步语义保证交接前的动作对对应接收方可见，却不会阻止一段错误代码继续写着它已经失去所有权的数组。

本例每个容器最多装三个整数。七个数对应三批，最后一批只有一个。结束标记使用同一种 Buffer，但要求 size 为零、end 为真；消费者仍然要实际参与最后一次交换，才能收到这条结束消息。

完整双线程程序

Java 17 即可运行，没有依赖。保存为 BufferExchange.java，执行 java BufferExchange.java。数据处理使用内存列表方便断言，缓冲容器则始终复用最初的两个对象。

import java.util.*;
import java.util.concurrent.*;

public class BufferExchange {
    static final class Buffer {
        final int[] values = new int[3];
        int size;
        boolean end;
        void clear() { size = 0; end = false; }
    }
    record Consumed(List<Integer> values, int batches) {}
    static Buffer swap(Exchanger<Buffer> exchanger, Buffer owned) throws Exception {
        return exchanger.exchange(owned, 1, TimeUnit.SECONDS);
    }
    static void produce(Exchanger<Buffer> exchanger, Buffer owned) throws Exception {
        for (int value = 1; value <= 7; value++) {
            owned.values[owned.size++] = value;
            if (owned.size == owned.values.length) {
                owned = swap(exchanger, owned);
                if (owned.size != 0 || owned.end) throw new IllegalStateException("empty buffer required");
            }
        }
        if (owned.size > 0) {
            owned = swap(exchanger, owned);
            if (owned.size != 0 || owned.end) throw new IllegalStateException("empty buffer required");
        }
        owned.end = true;
        swap(exchanger, owned); // End is a real exchanged message.
    }
    static Consumed consume(Exchanger<Buffer> exchanger, Buffer owned) throws Exception {
        List<Integer> values = new ArrayList<>();
        int batches = 0;
        for (;;) {
            owned = swap(exchanger, owned);
            if (owned.end) return new Consumed(List.copyOf(values), batches);
            if (owned.size <= 0 || owned.size > owned.values.length)
                throw new IllegalStateException("invalid data batch");
            for (int i = 0; i < owned.size; i++) values.add(owned.values[i]);
            batches++;
            owned.clear();
        }
    }
    public static void main(String[] args) throws Exception {
        Exchanger<Buffer> exchanger = new Exchanger<>();
        Buffer first = new Buffer(), second = new Buffer();
        ExecutorService executor = Executors.newFixedThreadPool(2);
        Future<?> producer = null;
        Future<Consumed> consumer = null;
        try {
            producer = executor.submit(() -> { produce(exchanger, first); return null; });
            consumer = executor.submit(() -> consume(exchanger, second));
            producer.get(4, TimeUnit.SECONDS);
            Consumed result = consumer.get(4, TimeUnit.SECONDS);
            if (!result.values().equals(List.of(1,2,3,4,5,6,7)) || result.batches() != 3)
                throw new AssertionError("lost or reordered batch");
            System.out.println("values=" + result.values());
            System.out.println("batches=" + result.batches());
            System.out.println("buffer objects=2");
            try {
                new Exchanger<Buffer>().exchange(new Buffer(), 40, TimeUnit.MILLISECONDS);
                throw new AssertionError("partner missing");
            } catch (TimeoutException expected) { System.out.println("single-side=timeout"); }
            Future<Boolean> interrupted = executor.submit(() -> {
                Thread.currentThread().interrupt();
                try { new Exchanger<Buffer>().exchange(new Buffer()); return false; }
                catch (InterruptedException expected) { return !Thread.currentThread().isInterrupted(); }
            });
            System.out.println("interruption-observed-and-cleared=" + interrupted.get(2, TimeUnit.SECONDS));
        } finally {
            if (producer != null) producer.cancel(true);
            if (consumer != null) consumer.cancel(true);
            executor.shutdownNow();
            if (!executor.awaitTermination(3, TimeUnit.SECONDS)) throw new IllegalStateException("workers still active");
        }
    }
}

values=[1, 2, 3, 4, 5, 6, 7]
batches=3
buffer objects=2
single-side=timeout
interruption-observed-and-cleared=true

三批分别包含 1、2、3，随后 4、5、6，最后 7。消费者把需要长期保存的整数值放进结果列表，再清空当前容器。如果改成把 values 数组直接存进结果，下一轮生产者会覆盖它，历史结果就会跟着变化。交换解决引用交接，不替调用方决定哪些数据需要复制。

这里的 buffer objects=2 指数据通路中的可复用缓冲对象；超时与中断测试各自创建独立实验对象，不参与三批数据。这一输出也不代表整个程序只分配两个对象：结果列表、Future、线程池等都有自己的生命周期。双缓冲减少的是特定数据容器的重复创建，不能据此编造总内存或吞吐收益。

尾批与结束消息为什么分开

如果七个数写完后直接退出，消费者可能已经处理了最后一批，却继续带着空容器等待下一次交换。生产者线程正常结束并不会让 Exchanger 自动返回一个 EOF。没有结束协议的双缓冲，常见表现就是数据看起来齐了，进程却一直活着。

程序先把未满的尾批交出去，再拿到一份空缓冲，设置 end 后进行最后交换。消费者见到 end 就停止，不再期待下一批；生产者也完成对应配对。两边交换次数一致，才构成正常关闭。即使数据量恰好是三的倍数，结束消息仍然必须发送，不能把“最后一批填满”当成流结束信号。

结束标记没有借用整数的某个特殊值，因此数据中出现零或负数不必与 EOF 争用编码。如果真实业务批次有类型、长度或校验信息，应把它们放到明确消息结构中；不要让消费者靠数组里恰好剩下几个默认零猜测有效范围。

也不要先清空自己的容器，再期待对方仍能读取旧数据。清空发生在消费者已经完成该批处理之后，它交出的才是下一轮可以复用的空容器。这个动作顺序就是所有权协议的一部分，比变量命名为 reusable 更有约束力。

一边离开以后，另一边不能永远等

swap 使用带一秒等待上限的 exchange，伙伴未到时抛 TimeoutException。单边实验故意不给对手，验证它确实退出。超时表示这一轮没有成功完成匹配，调用方不能继续把原对象当成已经交付，也不能假装整个批次成功结束。

若超时后自动重试，需要知道另一边此时处于哪一轮。两条线程可能因为各自错误进入不同的消息次序，盲目重复调用 exchange 会把新旧轮次配在一起。实际协议可以携带批次编号并统一结束失败状态；不是每个等待超时都适合从原地无限重试。

主线程通过 Future 观察任务失败，并在 finally 取消双方。已经发生的处理不会因此撤回，若消费者写了外部系统，重跑仍要考虑幂等或按批次核对。Exchanger 没有替外部副作用提供事务能力，这个事实与引用交接本身成功不矛盾。

所有等待都有限，但仍不保证任意代码能被一秒停止。消费者若在外部 I/O 中不响应中断，另一边等待超时只能终止自己的交换动作；关闭线程池还需要任务自身合作。示例在有限等待后确认线程终止，若没有终止则明确失败，不让后台线程静默遗留。

线程池也必须允许两个角色同时进入协议。若改成单线程执行器，生产者先占住唯一工作线程并等待交换，消费者还在队列里，永远没有机会到达。带超时的样例会失败退出，但超时只是暴露配置错误，不会凭空创造第二个执行位置。把互相等待的长期任务放进共享小线程池时，也要核对其他任务占用后是否仍有足够参与者。

同样不要在持有一把对方需要的业务锁时等待交换。另一方可能正等这把锁来处理或清空容器，而持锁方又等它到达交换点，形成跨机制的循环等待。最小程序只在各自独占缓冲区时处理整数，没有额外外部锁；接入真实处理阶段时，应把这项前提重新检查，而不是只看到使用了并发工具类就默认没有死锁。

中断会清除标志，处理方式由调用层决定

程序最后让任务先带上中断状态，再进入 exchange，收到 InterruptedException，并验证中断标志已清除。这是接口的规定行为，不代表中断请求应该被业务忽略。样例用它完成断言；通用任务若捕获后无法继续向上抛，应通常恢复中断状态并停止当前工作。

不要在 InterruptedException 后无条件再次进入同一交换循环。伙伴可能已经退出，当前批次的所有权与消息序号也可能需要重新核对。把中断当作“偶尔失败的一次调用”，会让取消请求反而制造新一轮等待。

相比队列，Exchanger 的双向交接很适合两个角色轮流交换空、满容器，双方速度会在交换点相互约束。若需求变成多个生产者、多种消费角色或允许大量积压，它的任意配对语义未必符合你的角色关系；多于两条线程也可以调用同一 Exchanger，但谁与谁配对不等于业务指定的固定搭档。

这里明确只有两个角色，数据通路不保存多份满缓冲，因此没有用一条无界队列掩盖慢消费者。真正的速度评估还要考虑填充、处理与等待各自成本；如果处理只占很少工作，同步本身也可能更显著。先验证消息次序和对象所有权，再决定是否值得引入双缓冲，比只看“少创建了几个数组”更能解释这段并发设计。