transfer 返回后任务还没成功：给 LinkedTransferQueue 消息单独确认

生产者调用transfer后正常返回，于是把任务状态改成“处理成功”。消费者其实刚刚take到对象，后面才开始校验、写入或调用远端。transfer等到的是接收，不是业务完成；一条消息从队列里消失，也不代表对应动作已经成功。

LinkedTransferQueue同时允许普通入队和等待直接交接，很容易让这两项合同混在一起。需要结果确认时，应在消息协议里放一个独立确认通道，由真正完成处理的一方给出结果或失败，而不是从队列方法的返回时点推断。

先交接，再等处理确认

下面在消息中放入CompletableFuture。它不是另一个线程，也不会自己执行处理，只保存本条消息的结果。消费者先接收第一条，再等一个门闩；生产者的transfer已经返回，确认Future仍然没有完成。

程序使用Java 17，保存为TransferReceipt.java后执行java TransferReceipt.java。第二条消息故意给出坏载荷，验证接收成功与处理失败可以同时成立。

import java.util.concurrent.*;
public class TransferReceipt {
    record Receipt(String id,String state) {}
    record Envelope(String id,String payload,CompletableFuture<Receipt> receipt) {}
    static Envelope message(String id,String payload) { return new Envelope(id,payload,new CompletableFuture<>()); }
    public static void main(String[] args) throws Exception {
        LinkedTransferQueue<Envelope> queue=new LinkedTransferQueue<>();
        ExecutorService workers=Executors.newFixedThreadPool(2);
        CountDownLatch processFirst=new CountDownLatch(1);
        Future<?> consumer=null, sending=null;
        try {
            consumer=workers.submit(()->{
                for(int i=0;i<2;i++) {
                    Envelope message=queue.poll(3,TimeUnit.SECONDS);
                    if(message==null) throw new IllegalStateException("message missing");
                    try {
                        if(message.id().equals("first") && !processFirst.await(3,TimeUnit.SECONDS))
                            throw new IllegalStateException("processing gate timeout");
                        if(message.payload().equals("bad")) throw new IllegalArgumentException("invalid payload");
                        message.receipt().complete(new Receipt(message.id(),"processed"));
                    } catch(InterruptedException e) {
                        message.receipt().completeExceptionally(e);
                        throw e;
                    } catch(RuntimeException e) { message.receipt().completeExceptionally(e); }
                }
                return null;
            });
            Envelope first=message("first","valid");
            sending=workers.submit(()->{queue.transfer(first);return null;});
            sending.get(3,TimeUnit.SECONDS);
            System.out.println("transfer returned, receipt done="+first.receipt().isDone());
            try { first.receipt().get(40,TimeUnit.MILLISECONDS);throw new AssertionError("receipt should wait"); }
            catch(TimeoutException expected) { System.out.println("receipt wait=timeout"); }
            processFirst.countDown();
            System.out.println("confirmed="+first.receipt().get(3,TimeUnit.SECONDS));
            Envelope second=message("second","bad");
            sending=workers.submit(()->{queue.transfer(second);return null;});
            sending.get(3,TimeUnit.SECONDS);
            try { second.receipt().get(3,TimeUnit.SECONDS);throw new AssertionError("failure expected"); }
            catch(ExecutionException expected) { System.out.println("processing failure="+expected.getCause().getMessage()); }
            consumer.get(3,TimeUnit.SECONDS);
            LinkedTransferQueue<String> buffered=new LinkedTransferQueue<>();
            buffered.offer("queued");
            System.out.println("offer without consumer size="+buffered.size());
            System.out.println("tryTransfer without consumer="+buffered.tryTransfer("not queued"));
        } finally {
            processFirst.countDown();
            if(sending!=null)sending.cancel(true);
            if(consumer!=null)consumer.cancel(true);
            workers.shutdownNow();
            if(!workers.awaitTermination(3,TimeUnit.SECONDS))throw new IllegalStateException("workers active");
        }
    }
}

transfer returned, receipt done=false
receipt wait=timeout
confirmed=Receipt[id=first, state=processed]
processing failure=invalid payload
offer without consumer size=1
tryTransfer without consumer=false

第一条输出的receipt done为false，来自明确的处理门闩，不能被解释为偶尔慢了一点。确认等待超时后，主线程才放开处理，最后拿到processed。第二条虽然也完成transfer，确认却返回ExecutionException，原因是消费者校验失败。

样例的processed只代表这段受控处理完成，没有远端服务或数据库。真实业务应当先确定需要确认的是校验完成、事务提交还是消息已持久化，再在那一个位置完成Future。若写入还没提交就先complete，确认通道只会用另一种形式重复原来的过早成功。

每条Envelope有独立id与receipt。若用一个共享Future处理多条消息，第一份结果就可能被所有调用者误读；若提交端只保留payload、不保留结果身份，异常出现时又可能找不到对应任务。消息身份与确认身份应在提交时建立稳定关联。

等确认超时，不会替你撤回已经交出的消息

get的40毫秒等待超时，没有取消消费者，也没有把消息从它手里取回。样例在超时后继续放开门闩并得到成功，正好说明“调用方不再愿意等”与“处理方已经停止”不同。需要取消时必须另行约定，且外部副作用可能无法撤回。

直接cancel确认Future也不等于中断消费者线程。消费者只是持有这份结果对象，并没有因此成为它的执行线程；后续complete可能返回false，而业务动作依旧发生。不能把一个被取消的结果容器，错误理解成整条处理链已经回滚。

调用方超时后重试，还需要稳定业务标识和幂等检查。旧消息可能已经处理但确认尚未观察到，新消息又重复执行；简单生成另一份Envelope会制造两个独立请求。结果Future只负责进程内确认，不提供跨进程重放保护或持久状态。

消费者异常时应让本条确认进入失败，不能只在自己的线程打印日志。否则发送方可能一直等到超时，失去实际失败原因。样例对非法载荷保留异常，InterruptedException则先完成失败再向外传播，不把关闭请求当作一条普通可继续消费的坏数据。

如果把这份确认封装成公共API，还应限制谁能完成它。示例各角色在同一个小程序里共享Envelope，方便观察；正式接口可以只向调用方暴露只读的CompletionStage视图，由处理组件保留完成权。否则调用方自己提前complete一个成功结果，就又绕过了我们刚建立的确认位置。结果容器可以等待，并不代表任何持有引用的人都应该拥有改写结果的权限。

offer 与 transfer 不能互相替换

最后两条输出使用没有消费者的独立队列：offer把元素存进去，size为一；tryTransfer找不到正在接收的一方，返回false，不把本次元素追加为待交付对象。普通入队与尝试立即交接，调用者要选择自己真正需要的等待政策。

LinkedTransferQueue是无界队列，普通offer不通过有限容量形成背压。生产快、处理慢时，积压对象可以持续增长。即使某些路径使用transfer限制单个生产者继续前进，其他代码绕过它改用offer，也可能把那份节奏约束打破；上限应在完整提交入口落实。

size还需要遍历，不是一个适合高频精确容量裁决的常量时间值；并发修改期间得到的数量也不等于原子快照。若需要限制全局在途任务，应使用独立计数或容量机制，并在失败、取消和确认路径完整归还资格，而不是先size再决定offer。

业务处理顺序也不能从FIFO接收直接推导。一个消费者接走慢任务，另一个接走快任务，后者可能先确认。要求按同一对象顺序提交结果时，需要该对象自己的串行或版本协议，队列只能约束它负责的交接阶段。

样例最后等待消费者Future，确认工作线程没有静默异常，再取消可能残留的发送任务并关闭线程池。若服务使用共享执行器，关闭权应留给它的拥有者；若消息确认可能长期留在Map里，还要设置保留期限。把交接、处理与确认分别命名以后，每一项超时和清理才能找到明确归属。