《Java 消息中间件·可靠消息与消费治理》2.2 幂等消费：唯一键、状态机与去重表

import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.atomic.LongAdder;
import java.util.function.LongSupplier;
import java.util.function.Supplier;

public final class IdempotentConsumer<V> {

    public enum Status { APPLIED, DUPLICATE }

    private enum State { PROCESSING, DONE }

    private static final class Slot<V> {
        State state = State.PROCESSING;
        V result; // 首次执行的结果，预留给需要返回值的调用方
        long expireAtMillis;
        boolean removed;
        final Object monitor = new Object();
    }

    public record Stats(long applied, long duplicate, long failed,
                        long rejected, long swept) { }

    private final ConcurrentHashMap<String, Slot<V>> table = new ConcurrentHashMap<>();
    private final LongSupplier clock;
    private final long ttlMillis;
    private final int maxEntries;
    private final LongAdder applied = new LongAdder();
    private final LongAdder duplicate = new LongAdder();
    private final LongAdder failed = new LongAdder();
    private final LongAdder rejected = new LongAdder();
    private final LongAdder swept = new LongAdder();

    public IdempotentConsumer(long ttlMillis, int maxEntries) {
        this(ttlMillis, maxEntries, System::currentTimeMillis);
    }

    /** clock 可注入，测试 TTL 时不用真的 sleep。 */
    public IdempotentConsumer(long ttlMillis, int maxEntries, LongSupplier clock) {
        if (ttlMillis <= 0) throw new IllegalArgumentException("ttlMillis must be > 0");
        if (maxEntries <= 0) throw new IllegalArgumentException("maxEntries must be > 0");
        this.ttlMillis = ttlMillis;
        this.maxEntries = maxEntries;
        this.clock = Objects.requireNonNull(clock);
    }

    public Status process(String key, Supplier<V> business) {
        Objects.requireNonNull(key, "key");
        Objects.requireNonNull(business, "business");
        while (true) {
            long now = clock.getAsLong();
            Slot<V> slot = table.get(key);
            if (slot != null && expired(slot, now)) {
                if (remove(key, slot)) {
                    slot = null;
                }
            }
            if (slot == null) {
                ensureCapacity();
                Slot<V> fresh = new Slot<>();
                Slot<V> previous = table.putIfAbsent(key, fresh);
                slot = previous == null ? fresh : previous;
            }
            synchronized (slot.monitor) {
                if (slot.removed) {
                    continue; // 与失败清理或 TTL 清理竞争，重来一次
                }
                if (slot.state == State.DONE) {
                    if (slot.expireAtMillis > clock.getAsLong()) {
                        duplicate.increment();
                        return Status.DUPLICATE;
                    }
                    removed(key, slot);
                    continue;
                }
                try {
                    V result = business.get();
                    slot.result = result;
                    slot.expireAtMillis = clock.getAsLong() + ttlMillis;
                    slot.state = State.DONE;
                    applied.increment();
                    return Status.APPLIED;
                } catch (RuntimeException | Error e) {
                    removed(key, slot);
                    failed.increment();
                    throw e;
                }
            }
        }
    }

    public int size() {
        return table.size();
    }

    public Stats stats() {
        return new Stats(applied.sum(), duplicate.sum(), failed.sum(),
                rejected.sum(), swept.sum());
    }

    /** 主动清理过期槽位，返回清理条数。 */
    public int sweep() {
        long now = clock.getAsLong();
        int removed = 0;
        for (Map.Entry<String, Slot<V>> entry : table.entrySet()) {
            Slot<V> slot = entry.getValue();
            synchronized (slot.monitor) {
                if (!slot.removed && expired(slot, now)) {
                    slot.removed = true;
                    if (table.remove(entry.getKey(), slot)) {
                        removed++;
                    }
                }
            }
        }
        swept.add(removed);
        return removed;
    }

    private boolean expired(Slot<V> slot, long now) {
        return slot.state == State.DONE && slot.expireAtMillis <= now;
    }

    private boolean remove(String key, Slot<V> slot) {
        synchronized (slot.monitor) {
            if (slot.removed) return false;
            slot.removed = true;
            return table.remove(key, slot);
        }
    }

    private void removed(String key, Slot<V> slot) {
        slot.removed = true;
        table.remove(key, slot);
    }

    private void ensureCapacity() {
        if (table.size() < maxEntries) return;
        sweep();
        if (table.size() >= maxEntries) {
            rejected.increment();
            throw new RejectedExecutionException(
                    "idempotent table full: size=" + table.size() + " max=" + maxEntries);
        }
    }
}

// 16 个线程同时投递同一个 key
CountDownLatch start = new CountDownLatch(1);
for (int i = 0; i < 16; i++) {
    pool.submit(() -> {
        start.await();
        Status s = consumer.process("loan-42", () -> {
            calls.incrementAndGet();
            sleep(20);
            return "ok";
        });
        if (s == Status.APPLIED) applied.incrementAndGet();
        else duplicate.incrementAndGet();
        return null;
    });
}
start.countDown();

single-key: status=APPLIED,DUPLICATE,DUPLICATE businessCalls=1 stats=Stats[applied=1, duplicate=2, failed=0, rejected=0, swept=0]
concurrent-same-key: threads=16 applied=1 duplicate=15 businessCalls=1 stats=Stats[applied=1, duplicate=15, failed=0, rejected=0, swept=0]
many-keys: keys=1000 deliveries=3000 businessCalls=1000 tableSize=1000 stats=Stats[applied=1000, duplicate=2000, failed=0, rejected=0, swept=0] elapsedMs=165 (单次运行，随调度波动，不作基准)
failure-retry: attempts=2 second=APPLIED stats=Stats[applied=1, duplicate=0, failed=1, rejected=0, swept=0] tableSize=1
ttl: first=APPLIED withinTtl=DUPLICATE afterTtl=APPLIED businessCalls=2 swept=1 tableSize=0 stats=Stats[applied=2, duplicate=1, failed=0, rejected=0, swept=1]
capacity: atLimit=1000 rejected=true swept=1000 reinsert=APPLIED stats=Stats[applied=1001, duplicate=0, failed=0, rejected=1, swept=1000]

1000 × 86400 × 30 = 2,592,000,000 条

enum LoanState { CREATED, APPROVED, RETURNED, CANCELLED }

static LoanState transition(LoanState current, LoanState target) {
    if (current == target) return current; // 目标态已达成，幂等返回
    return switch (current) {
        case CREATED -> {
            if (target == LoanState.APPROVED || target == LoanState.CANCELLED) yield target;
            throw new IllegalStateException(current + " -> " + target);
        }
        case APPROVED -> {
            if (target == LoanState.RETURNED) yield target;
            throw new IllegalStateException(current + " -> " + target);
        }
        case RETURNED, CANCELLED -> throw new IllegalStateException(current + " -> " + target);
    };
}

state-machine: CREATED->APPROVED=APPROVED repeat=APPROVED APPROVED->CREATED=IllegalStateException