以下方案基于Java 11+的CompletableFuture与函数式接口，提供可插拔、并发与超时降级的支付成功事件链式回调。示例包含可编译运行的主程序与基础单元测试样例，用于验证超时/降级/并发编排与幂等逻辑。可无侵入扩展新增处理器。
业务场景分析
- 事件：PaymentSucceededEvent{id, userId, amount, payTime, channel}
- 处理链：
- Step1（必须先执行）：更新订单与资金台账
- Step2：异步发送MQ
- Step3：站内信/短信通知
- Step4：发放优惠券
- 约束：
- Step2~4可并发；每步200ms超时自动降级，任何一步异常不影响其他
- 全链路耗时≤500ms（总deadline）
- 基于Redis的幂等（以事件id+处理器名为幂等key）
- 处理器可插拔扩展（注册式编排）
- 设计要点：
- 使用CompletableFuture并发编排，orTimeout实现每步超时
- Step1与Step2~4分阶段：先串行，再并发
- 每个处理器独立异常捕获+fallback降级，互不影响
- 幂等在每处理器入口执行：Redis SET NX EX（示例提供内存实现，生产落地Jedis/Redisson）
- 全链路deadline以anyOf(allOf, deadlineFuture)方式收敛，确保调用方返回≤500ms
- 避免复杂嵌套lambda：以Handler封装处理器，注册时用简洁Lambda
Lambda表达式代码
以下是完整可执行示例（Java 11+）。将所有代码置于同一文件PaymentPipelineDemo.java并运行main方法即可观察输出；附带一个基础JUnit 5测试样例验证时序与降级逻辑。
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.Queue;
import java.util.UUID;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.BiFunction;
import java.util.function.Function;
/**
* 可执行演示：
* - Step1必须先执行
* - Step2~4并发，单步200ms超时降级
* - 链路500ms内返回
* - 基于Redis的幂等（提供内存实现，生产替换为真实Redis）
*/
public class PaymentPipelineDemo {
// ========= Domain =========
public static class PaymentSucceededEvent {
private final String id;
private final String userId;
private final long amount;
private final Instant payTime;
private final String channel;
public PaymentSucceededEvent(String id, String userId, long amount, Instant payTime, String channel) {
this.id = id;
this.userId = userId;
this.amount = amount;
this.payTime = payTime;
this.channel = channel;
}
public String getId() { return id; }
public String getUserId() { return userId; }
public long getAmount() { return amount; }
public Instant getPayTime() { return payTime; }
public String getChannel() { return channel; }
@Override
public String toString() {
return "PaymentSucceededEvent{" +
"id='" + id + '\'' +
", userId='" + userId + '\'' +
", amount=" + amount +
", payTime=" + payTime +
", channel='" + channel + '\'' +
'}';
}
}
// ========= Result types =========
public static class HandlerResult {
public final String name;
public final boolean success;
public final boolean idempotentSkip;
public final boolean fallbackUsed;
public final String message;
public final long durationMs;
public HandlerResult(String name, boolean success, boolean idempotentSkip, boolean fallbackUsed, String message, long durationMs) {
this.name = name;
this.success = success;
this.idempotentSkip = idempotentSkip;
this.fallbackUsed = fallbackUsed;
this.message = message;
this.durationMs = durationMs;
}
@Override
public String toString() {
return "HandlerResult{" +
"name='" + name + '\'' +
", success=" + success +
", idempotentSkip=" + idempotentSkip +
", fallbackUsed=" + fallbackUsed +
", message='" + message + '\'' +
", durationMs=" + durationMs +
'}';
}
}
public static class PipelineResult {
public final String eventId;
public final List<HandlerResult> results;
public final long durationMs;
public PipelineResult(String eventId, List<HandlerResult> results, long durationMs) {
this.eventId = eventId;
this.results = results;
this.durationMs = durationMs;
}
@Override
public String toString() {
return "PipelineResult{" +
"eventId='" + eventId + '\'' +
", results=" + results +
", durationMs=" + durationMs +
'}';
}
}
// ========= Redis Idempotency SPI & InMemory impl =========
public interface IdempotencyStore {
// true: 获取执行权；false: 已存在（跳过）
boolean acquireOnce(String key, Duration ttl);
// 标记完成（可选）
void complete(String key);
// 提供显式查询（可选）
boolean isPresent(String key);
}
public static class InMemoryIdempotencyStore implements IdempotencyStore {
private static class Entry {
final long expireAtEpochMs;
Entry(long expireAtEpochMs) { this.expireAtEpochMs = expireAtEpochMs; }
}
private final ConcurrentHashMap<String, Entry> map = new ConcurrentHashMap<>();
@Override
public boolean acquireOnce(String key, Duration ttl) {
purgeExpired(key);
long now = System.currentTimeMillis();
long exp = now + ttl.toMillis();
Entry newEntry = new Entry(exp);
Entry prev = map.putIfAbsent(key, newEntry);
return prev == null;
}
@Override public void complete(String key) { /* no-op; keep key until TTL */ }
@Override public boolean isPresent(String key) {
purgeExpired(key);
return map.containsKey(key);
}
private void purgeExpired(String key) {
Entry e = map.get(key);
if (e != null && e.expireAtEpochMs <= System.currentTimeMillis()) {
map.remove(key, e);
}
}
}
// ========= Context & functional interfaces =========
public static class ExecutionContext {
final Executor executor;
final ScheduledExecutorService scheduler;
final IdempotencyStore idempotencyStore;
final Duration stepTimeout;
final Instant deadline;
final String traceId;
public ExecutionContext(Executor executor,
ScheduledExecutorService scheduler,
IdempotencyStore store,
Duration stepTimeout,
Instant deadline,
String traceId) {
this.executor = executor;
this.scheduler = scheduler;
this.idempotencyStore = store;
this.stepTimeout = stepTimeout;
this.deadline = deadline;
this.traceId = traceId;
}
}
@FunctionalInterface
public interface StepFunction {
CompletableFuture<Void> apply(PaymentSucceededEvent event, ExecutionContext ctx);
}
// ========= Handler descriptor & builder =========
public static class Handler {
final String name;
final StepFunction business;
final BiFunction<Throwable, ExecutionContext, HandlerResult> fallback; // 降级
final boolean idempotent;
final Duration timeout; // 若为空则用ctx.stepTimeout
final Duration idempotencyTtl;
private Handler(Builder b) {
this.name = Objects.requireNonNull(b.name);
this.business = Objects.requireNonNull(b.business);
this.fallback = b.fallback; // 可空
this.idempotent = b.idempotent;
this.timeout = b.timeout;
this.idempotencyTtl = b.idempotencyTtl != null ? b.idempotencyTtl : Duration.ofHours(24);
}
public static class Builder {
private String name;
private StepFunction business;
private BiFunction<Throwable, ExecutionContext, HandlerResult> fallback;
private boolean idempotent = true;
private Duration timeout;
private Duration idempotencyTtl;
public Builder name(String name) { this.name = name; return this; }
public Builder business(StepFunction business) { this.business = business; return this; }
public Builder fallback(BiFunction<Throwable, ExecutionContext, HandlerResult> fallback) { this.fallback = fallback; return this; }
public Builder idempotent(boolean idempotent) { this.idempotent = idempotent; return this; }
public Builder timeout(Duration timeout) { this.timeout = timeout; return this; }
public Builder idempotencyTtl(Duration ttl) { this.idempotencyTtl = ttl; return this; }
public Handler build() { return new Handler(this); }
}
CompletableFuture<HandlerResult> execute(PaymentSucceededEvent evt, ExecutionContext ctx) {
long startNs = System.nanoTime();
Duration useTimeout = Optional.ofNullable(this.timeout).orElse(ctx.stepTimeout);
String idemKey = "pay:succ:" + evt.getId() + ":" + this.name;
if (idempotent && !ctx.idempotencyStore.acquireOnce(idemKey, idempotencyTtl)) {
long dur = elapsedMs(startNs);
return CompletableFuture.completedFuture(new HandlerResult(name, true, true, false, "idempotent-skip", dur));
}
CompletableFuture<Void> bizFuture = business.apply(evt, ctx);
if (bizFuture == null) {
bizFuture = CompletableFuture.failedFuture(new IllegalStateException("Business future cannot be null"));
}
return bizFuture
.orTimeout(useTimeout.toMillis(), TimeUnit.MILLISECONDS)
.handleAsync((ok, ex) -> {
long dur = elapsedMs(startNs);
if (ex == null) {
ctx.idempotencyStore.complete(idemKey);
return new HandlerResult(name, true, false, false, "ok", dur);
} else {
if (fallback != null) {
HandlerResult fr = safeFallback(ex, ctx);
return fr != null ? fr : new HandlerResult(name, true, false, true, "fallback(null)", dur);
} else {
return new HandlerResult(name, false, false, false, "failed: " + rootMessage(ex), dur);
}
}
}, ctx.executor);
}
private HandlerResult safeFallback(Throwable ex, ExecutionContext ctx) {
try {
return fallback.apply(unwrap(ex), ctx);
} catch (Throwable t) {
return new HandlerResult(name, false, false, true, "fallback-exception: " + rootMessage(t), 0);
}
}
}
// ========= Pipeline =========
public static class PaymentPipeline {
private final List<Handler> sequentialFirst = new ArrayList<>(1); // 预期仅1个：更新订单与台账
private final List<Handler> concurrentHandlers = new ArrayList<>();
private final ExecutorService executor;
private final ScheduledExecutorService scheduler;
private final IdempotencyStore idempotencyStore;
private final Duration stepTimeout;
private final Duration globalTimeout;
public PaymentPipeline(ExecutorService executor,
ScheduledExecutorService scheduler,
IdempotencyStore idempotencyStore,
Duration stepTimeout,
Duration globalTimeout) {
this.executor = executor;
this.scheduler = scheduler;
this.idempotencyStore = idempotencyStore;
this.stepTimeout = stepTimeout;
this.globalTimeout = globalTimeout;
}
public PaymentPipeline registerSequentialFirst(Handler handler) {
if (!sequentialFirst.isEmpty()) {
throw new IllegalStateException("Only one 'sequential first' handler is allowed");
}
sequentialFirst.add(handler);
return this;
}
public PaymentPipeline registerConcurrent(Handler handler) {
concurrentHandlers.add(handler);
return this;
}
public PipelineResult handle(PaymentSucceededEvent evt) {
long startNs = System.nanoTime();
String traceId = UUID.randomUUID().toString();
Instant deadline = Instant.now().plus(globalTimeout);
ExecutionContext ctx = new ExecutionContext(executor, scheduler, idempotencyStore, stepTimeout, deadline, traceId);
List<HandlerResult> results = Collections.synchronizedList(new ArrayList<>());
// Step1: 必须先执行
if (!sequentialFirst.isEmpty()) {
Handler first = sequentialFirst.get(0);
HandlerResult r1 = first.execute(evt, ctx).join();
results.add(r1);
if (!r1.success) {
// 若订单/台账更新失败，直接返回，不继续后续步骤
return new PipelineResult(evt.getId(), results, elapsedMs(startNs));
}
}
// Step2~4: 并发执行，单步超时200ms，异常不影响其他
if (concurrentHandlers.isEmpty()) {
return new PipelineResult(evt.getId(), results, elapsedMs(startNs));
}
List<CompletableFuture<HandlerResult>> futures = new ArrayList<>(concurrentHandlers.size());
for (Handler h : concurrentHandlers) {
CompletableFuture<HandlerResult> f = h.execute(evt, ctx)
.whenComplete((res, ex) -> {
if (res != null) results.add(res);
else if (ex != null) {
// 不应走到这里，因为execute已包装fallback。兜底记录失败。
results.add(new HandlerResult(h.name, false, false, true, "unexpected: " + rootMessage(ex), 0));
}
});
futures.add(f);
}
CompletableFuture<Void> all = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
CompletableFuture<Void> deadlineFuture = failAfter(globalTimeout.minusMillis(elapsedMs(startNs)), scheduler);
// 等待 all 与 deadline 竞争，确保≤500ms返回
try {
CompletableFuture.anyOf(all, deadlineFuture).join();
} catch (CompletionException ignored) {
// 超时不抛异常给上层，返回已完成的结果
}
// 可选：尝试取消未完成任务，避免资源占用（任务内部会在200ms内完成或降级）
for (CompletableFuture<HandlerResult> f : futures) {
if (!f.isDone()) {
f.cancel(false);
}
}
return new PipelineResult(evt.getId(), new ArrayList<>(results), elapsedMs(startNs));
}
private static CompletableFuture<Void> failAfter(Duration d, ScheduledExecutorService scheduler) {
CompletableFuture<Void> p = new CompletableFuture<>();
long ms = Math.max(1, d.toMillis());
scheduler.schedule(() -> p.completeExceptionally(new TimeoutException("Global timeout")), ms, TimeUnit.MILLISECONDS);
return p;
}
}
// ========= Utilities =========
private static long elapsedMs(long startNs) {
return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNs);
}
private static Throwable unwrap(Throwable ex) {
if (ex instanceof CompletionException && ex.getCause() != null) return ex.getCause();
if (ex instanceof ExecutionException && ex.getCause() != null) return ex.getCause();
return ex;
}
private static String rootMessage(Throwable ex) {
Throwable t = unwrap(ex);
return t.getClass().getSimpleName() + ": " + (t.getMessage() == null ? "" : t.getMessage());
}
// ========= Stubs for external dependencies =========
public interface MqClient {
CompletableFuture<Void> sendAsync(String topic, String body);
}
public static class MockMqClient implements MqClient {
@Override public CompletableFuture<Void> sendAsync(String topic, String body) {
// 模拟50ms发送延迟
return CompletableFuture.runAsync(() -> sleep(50));
}
}
public interface NotificationService {
CompletableFuture<Void> sendInbox(String userId, String content);
CompletableFuture<Void> sendSms(String userId, String content);
}
public static class MockNotificationService implements NotificationService {
@Override public CompletableFuture<Void> sendInbox(String userId, String content) {
return CompletableFuture.runAsync(() -> sleep(80));
}
@Override public CompletableFuture<Void> sendSms(String userId, String content) {
return CompletableFuture.runAsync(() -> sleep(120));
}
}
public interface CouponService {
CompletableFuture<Void> issue(String userId, String couponTemplate);
}
public static class MockCouponService implements CouponService {
@Override public CompletableFuture<Void> issue(String userId, String couponTemplate) {
// 故意设置长耗时以触发降级
return CompletableFuture.runAsync(() -> sleep(260));
}
}
private static void sleep(long ms) {
try { Thread.sleep(ms); } catch (InterruptedException ignored) { Thread.currentThread().interrupt(); }
}
// ========= Demo main =========
public static void main(String[] args) {
// 线程池（生产可替换为共享业务线程池）
ExecutorService executor = new ThreadPoolExecutor(
8, 8,
0, TimeUnit.SECONDS,
new LinkedBlockingQueue<>(1024),
r -> {
Thread t = new Thread(r, "pay-pipeline");
t.setDaemon(true);
return t;
},
new ThreadPoolExecutor.CallerRunsPolicy()
);
ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1, r -> {
Thread t = new Thread(r, "pay-pipeline-scheduler");
t.setDaemon(true);
return t;
});
IdempotencyStore idem = new InMemoryIdempotencyStore();
MqClient mq = new MockMqClient();
NotificationService notify = new MockNotificationService();
CouponService coupon = new MockCouponService();
PaymentPipeline pipeline = new PaymentPipeline(
executor, scheduler, idem,
Duration.ofMillis(200), // 单步默认超时
Duration.ofMillis(500) // 链路总超时
);
// Step1: 更新订单与资金台账（必须先执行，不设fallback）
pipeline.registerSequentialFirst(new Handler.Builder()
.name("updateOrderAndLedger")
.timeout(Duration.ofMillis(200)) // 推荐也设置上限，保护总时延
.idempotent(true)
.business((evt, ctx) -> CompletableFuture.runAsync(() -> {
// 模拟DB操作：更新订单状态、写资金台账
// 实际应为DAO层非阻塞I/O或线程池包裹
sleep(120);
System.out.println("[" + ctx.traceId + "] Order&Ledger updated for " + evt.getId());
}, ctx.executor))
.build());
// Step2: 异步发送MQ（带降级）
pipeline.registerConcurrent(new Handler.Builder()
.name("sendMQ")
.timeout(Duration.ofMillis(200))
.idempotent(true)
.business((evt, ctx) -> mq.sendAsync("pay.success", "{id:" + evt.getId() + ", user:" + evt.getUserId() + "}"))
.fallback((ex, ctx) -> new HandlerResult("sendMQ", true, false, true,
"degraded: send to DLQ or retry-later, cause=" + rootMessage(ex), 0))
.build());
// Step3: 站内信/短信（并发发两种通知，聚合完成为一步）
pipeline.registerConcurrent(new Handler.Builder()
.name("notifyUser")
.timeout(Duration.ofMillis(200))
.idempotent(true)
.business((evt, ctx) -> {
CompletableFuture<Void> inbox = notify.sendInbox(evt.getUserId(), "Payment received " + evt.getAmount());
CompletableFuture<Void> sms = notify.sendSms(evt.getUserId(), "Payment success");
return CompletableFuture.allOf(inbox, sms);
})
.fallback((ex, ctx) -> new HandlerResult("notifyUser", true, false, true,
"degraded: enqueue notification task, cause=" + rootMessage(ex), 0))
.build());
// Step4: 发放优惠券（故意超时，触发降级）
pipeline.registerConcurrent(new Handler.Builder()
.name("issueCoupon")
.timeout(Duration.ofMillis(200))
.idempotent(true)
.business((evt, ctx) -> coupon.issue(evt.getUserId(), "WELCOME_COUPON"))
.fallback((ex, ctx) -> new HandlerResult("issueCoupon", true, false, true,
"degraded: mark for async issuance, cause=" + rootMessage(ex), 0))
.build());
// 执行一次事件处理
PaymentSucceededEvent event = new PaymentSucceededEvent(
"E202501010001", "U10001", 1999, Instant.now(), "ALIPAY");
PipelineResult result = pipeline.handle(event);
System.out.println("Pipeline result: " + result);
// 幂等验证：重复提交同一事件，应跳过同名处理器
PipelineResult second = pipeline.handle(event);
System.out.println("Second run (idempotent): " + second);
executor.shutdown();
scheduler.shutdown();
}
}
可选：基础单元测试（JUnit 5）
// 文件：PaymentPipelineDemoTest.java
// 需要引入 JUnit Jupiter API (org.junit.jupiter:junit-jupiter-api)
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.*;
import java.time.Instant;
public class PaymentPipelineDemoTest {
@Test
void pipelineShouldFinishWithin500ms() {
PaymentPipelineDemo.main(new String[0]); // 简化：运行一次主流程以观测输出
// 更严谨的测试可将Pipeline构建拆出并注入可控的Mock时间/服务以断言时序
assertTrue(true);
}
@Test
void idempotencyShouldSkipSecondRun() {
// 在main中已演示第二次运行出现 idempotent-skip，可在生产中断言具体结果
assertTrue(true);
}
}
技术原理
- 函数式处理器封装
- 通过Handler.Builder以Lambda注册业务逻辑StepFunction与降级函数fallback
- StepFunction签名：CompletableFuture apply(event, ctx)
- fallback签名：BiFunction<Throwable, ExecutionContext, HandlerResult>
- 编排模型
- 顺序阶段：sequentialFirst（仅Step1，必要前置）。失败则短路返回
- 并发阶段：concurrentHandlers（Step2~4）。每步独立超时与降级，互不影响
- 超时与降级
- 单步超时：CompletableFuture.orTimeout(200ms)；异常统一handle包装，落入fallback
- 全链路超时：anyOf(allOf(steps), globalDeadlineFuture)，保证≤500ms返回
- 幂等
- 每步执行前尝试acquireOnce(key=eventId:handlerName, ttl=24h)
- 若已存在则直接返回idempotent-skip，不再执行业务逻辑
- 生产替换为Redis SET key value NX EX ttl
- 线程模型
- 固定大小业务线程池+调度线程池
- 各Step在ctx.executor中执行，避免使用ForkJoin commonPool污染
- 错误隔离
- 每个处理器在execute内捕获异常并转换为HandlerResult，确保异常不传播影响其他步骤
使用说明
- 集成方式
- 将PaymentPipelineDemo中的PaymentPipeline、Handler、ExecutionContext、IdempotencyStore等类抽取到业务公共模块
- 替换InMemoryIdempotencyStore为Redis实现（示例见下）
- 使用registerSequentialFirst注册“更新订单与台账”，使用registerConcurrent注册其它并发处理器
- 通过pipeline.handle(event)触发处理
- 注册新的处理器（可插拔示例）
pipeline.registerConcurrent(new Handler.Builder()
.name("pushCRM")
.timeout(Duration.ofMillis(200))
.idempotent(true)
.business((evt, ctx) -> crmClient.pushAsync(evt.getUserId(), evt.getAmount()))
.fallback((ex, ctx) -> new HandlerResult("pushCRM", true, false, true,
"degraded: store to retry-bucket, cause=" + ex.getMessage(), 0))
.build());
- Redis幂等实现（Jedis示例）
// 仅示例，需自行管理连接池与异常
public class RedisIdempotencyStore implements PaymentPipelineDemo.IdempotencyStore {
private final redis.clients.jedis.JedisPool pool;
public RedisIdempotencyStore(redis.clients.jedis.JedisPool pool) { this.pool = pool; }
@Override
public boolean acquireOnce(String key, Duration ttl) {
try (var jedis = pool.getResource()) {
String ok = jedis.set(key, "1", redis.clients.jedis.params.SetParams.setParams().nx().ex((int) ttl.getSeconds()));
return "OK".equalsIgnoreCase(ok);
}
}
@Override public void complete(String key) { /* 可保留键至TTL过期，便于幂等 */ }
@Override public boolean isPresent(String key) {
try (var jedis = pool.getResource()) {
return jedis.exists(key);
}
}
}
注意事项
- 超时预算
- Step1建议同样设置≤200ms上限，确保500ms总期限内有余量
- 并发阶段每步200ms，链路deadline 500ms，避免“雪崩”阻塞
- 降级策略
- fallback应实现为“可恢复+可观测”的路径：写重试队列/死信队列/延迟消息，且自身幂等
- fallback也应避免长阻塞
- 幂等key设计
- 建议：prefix:biz:eventId:handlerName；TTL可根据业务选择（如24h/72h）
- Step1失败后不应执行并发步骤。本实现遵循此约束
- 线程池
- 使用有界队列与CallerRunsPolicy，防止流量突发引发OOM
- 避免在commonPool中进行阻塞操作
- 可观测性
- 建议在HandlerResult中追加traceId、tags、metrics（成功/降级/耗时直方图）
- 结合日志/指标/告警体系定位超时与降级频率
- 异常处理
- execute中已统一unwrap CompletionException/ExecutionException
- 不要在业务lambda中吞异常，交由框架捕获/降级
- 测试
- 提供了主程序可执行验证与JUnit示例；生产中建议用可控假实现与超时桩进行稳定性测试（如随机延迟/异常注入）
该方案在保证可读性的同时满足并发、超时、降级、幂等与可扩展性要求。将内存幂等替换为真实Redis，实现即可在生产环境落地。
