16 KiB
11. 跨域协作与聚合
决策
跨 domain 的读走契约模块（-contract）里的接口，跨 domain 的写/状态联动走 Spring ApplicationEvent + @TransactionalEventListener，暂不引入消息中间件。bff-orchestration / workbench 的并行聚合用一个专用的有界线程池，带上下文传播和整体超时预算，按 tile 局部降级。
这一篇填的是原来整套文档最大的一个空白：04-security-auth.md 里写了"切店 → 通知 webview-ticket 失效"走事件，但事件机制本身从来没有被定义过。
一、跨域读：契约模块
模块结构和约束见 01-project-structure.md，这里只讲使用规则。
// domains/identity-store-contract/src/main/kotlin/.../identitystore/contract/StoreQueryService.kt
interface StoreQueryService {
fun listStoresByUserId(userId: Long): List<StoreInfo>
fun findStore(storeId: Long): StoreInfo?
}
data class StoreInfo(
val storeId: Long,
val name: String,
val code: String,
)
// domains/identity-store/src/main/kotlin/.../identitystore/application/StoreQueryServiceImpl.kt
@Service
class StoreQueryServiceImpl(private val storeRepository: StoreRepository) : StoreQueryService {
override fun listStoresByUserId(userId: Long): List<StoreInfo> =
storeRepository.findStoresByUserId(userId).map { StoreInfo(it.id, it.name, it.code) }
}
// domains/workbench/src/main/kotlin/.../workbench/application/WorkbenchAppService.kt
@Service
class WorkbenchAppService(
private val storeQueryService: StoreQueryService, // 注入的是契约里的接口，不是 identity-store 的实现类
) { ... }
契约设计的四条规则
- 契约模型是独立的数据结构，不是 Entity 的别名。
StoreInfo只含调用方真正需要的字段。让它跟着StoreEntity一起长，等于把内部表结构变成了对外承诺，改一个列名就要动三个 domain。 - 契约只承诺调用方需要的最小能力。不要一上来就写
findAll()、update()——契约里出现的每个方法都是未来的约束。 - 契约变更要向后兼容。加方法、加可空字段没问题；删方法、改语义要先确认所有调用方，跟对客户端的 API 一个待遇。
- 契约模块不含实现、不依赖 Spring Web/JPA（ArchUnit 会检查，见 10-testing.md）。
什么时候不该用契约，而应该重新划边界
如果 workbench 要用 identity-store 的契约方法超过五六个，甚至开始要求对方加"给我拼好这个结构"的定制方法，那说明边界划错了——这块逻辑本来就该在一边，或者本来就该单独成域。契约模块变厚是个明确的设计告警，不要靠往里加方法来消化它。
二、跨域写：领域事件
为什么写操作不走契约接口
workbench 直接调 identityStore.doSomething() 意味着：workbench 要知道 identity-store 内部该做什么，identity-store 的事务边界被外部方法调用拉长，而且以后每多一个关心"切店"的域，就要在切店逻辑里多加一行调用——切店代码变成一个不断膨胀的通知中心。
事件反转了这个依赖方向：发布方不知道谁在听。切店只管发一个"门店切换了"的事实，谁关心谁自己订阅。
事件定义放在契约模块
// domains/identity-store-contract/src/main/kotlin/.../identitystore/contract/StoreSwitchedEvent.kt
data class StoreSwitchedEvent(
val userId: Long,
val fromStoreId: Long?,
val toStoreId: Long,
val occurredAt: Instant,
)
事件类型必须放在契约模块，否则订阅方要 import 发布方的内部类型，边界又破了。
发布：在事务内发布
// domains/identity-store/src/main/kotlin/.../identitystore/application/StoreSwitchAppService.kt
@Service
class StoreSwitchAppService(
private val events: ApplicationEventPublisher,
private val clock: Clock,
) {
@Transactional
fun switchStore(userId: Long, targetStoreId: Long): StoreContext {
val store = findAccessibleStore(userId, targetStoreId)
?: throw BusinessException(ErrorCode.STORE_NOT_ACCESSIBLE, "无权访问该门店", HttpStatus.FORBIDDEN)
// ... 更新当前门店、重新签发 access token（见 04-security-auth.md）
events.publishEvent(StoreSwitchedEvent(userId, currentStoreId, targetStoreId, clock.instant()))
return context
}
}
订阅：@TransactionalEventListener
// domains/webview-ticket/src/main/kotlin/.../webviewticket/application/StoreSwitchedListener.kt
@Component
class StoreSwitchedListener(private val ticketRepository: WebviewTicketRepository) {
private val log = LoggerFactory.getLogger(javaClass)
@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
@Transactional(propagation = Propagation.REQUIRES_NEW)
fun onStoreSwitched(event: StoreSwitchedEvent) {
runCatching { ticketRepository.revokeActiveTickets(event.userId) }
.onFailure { log.error("切店后作废 webview 票据失败 userId={}", event.userId, it) }
}
}
这段代码里每个注解和写法都在解决一个具体问题：
| 写法 | 解决什么 |
|---|---|
@TransactionalEventListener(AFTER_COMMIT) |
普通 @EventListener 是同步、在同一事务内执行的。用它意味着"切店事务回滚了，但票据已经作废"这种不一致；AFTER_COMMIT 保证只在主事务真正提交后才触发 |
@Transactional(REQUIRES_NEW) |
AFTER_COMMIT 阶段原事务已经提交，此时没有活跃事务。不开新事务的话，监听器里的写操作要么报错要么自动提交，行为不可控 |
runCatching + 记日志 |
监听器抛异常不会回滚主事务（主事务已提交），只会让这次副作用静默丢失。必须显式捕获并留下可排查的日志 |
同步还是异步
默认同步（AFTER_COMMIT 但仍在同一线程）。理由：同步下 traceId、@RequestScope 的门店上下文都还在，出问题能直接顺着日志查下去；异步则要额外处理上下文传播，而目前这些副作用（作废票据、清缓存）都很轻，没必要付这个复杂度。
只有当某个监听器确实耗时（比如要调外部系统）时才加 @Async，并且必须：指定专用线程池（不要用默认的 SimpleAsyncTaskExecutor，它每次新建线程且无上界）、加上下文传播的 TaskDecorator（见下文第三节）、想清楚失败后怎么办。
事件的可靠性边界（必须如实认识）
ApplicationEvent 是进程内、内存中的：应用在事件发出后、监听器执行前崩溃，这个事件就永久丢了，没有重试、没有补偿。
所以这套机制只能用于"丢了不致命"的副作用——作废一张票据（下次换票时本来也会重新校验）、清一个缓存、记一条审计日志。不能用于：扣款、发货、任何丢了会造成数据不一致且无法自愈的操作。
在当前这套系统里，跨域事件的用途就是切店后作废票据这一类，符合这个边界。
什么时候升级到消息中间件
出现下面任意一条，就该引入 RabbitMQ/Kafka + 事务性发件箱（transactional outbox），而不是继续给 ApplicationEvent 打补丁：
- 事件消费失败需要自动重试，或者需要死信队列；
- 事件丢失会造成业务上的资金/库存不一致；
- 消费方被拆成了独立进程（模块化单体拆微服务时的必然结果）；
- 需要一个事件被多个消费组各自独立消费，且各自有独立的消费进度。
事务性发件箱的做法（记在这里备查，现在不实施）：在业务事务里往 outbox 表插一条记录（与业务写在同一个事务，天然原子），另有一个轮询任务把 outbox 里的记录投递到消息中间件并标记已发送。它解决的是"写库成功但发消息失败"这个用 AFTER_COMMIT 无论如何都消除不掉的窗口。
现在不做的理由很直接：引入中间件意味着多一套需要部署、监控、排障的基础设施，而当前唯一的跨域事件场景丢了也不致命。等到有第一个"丢了会出事"的事件时再做，那时候需求也更清楚。
三、聚合：bff-orchestration 与 workbench 的并行 fan-out
首页要同时拉采购、保修、门店信息等多个 tile（架构图 Flow 3），串行调用意味着总耗时是各下游耗时之和。同步栈下必须显式用线程池做并行。
专用线程池
// domains/workbench/src/main/kotlin/.../workbench/infrastructure/config/WorkbenchExecutorConfig.kt
// 模块内的 @Configuration 一律放 infrastructure/config/（见 01-project-structure.md 的脚手架），
// 不要散在模块根包下——那样它既不属于任何一层，ArchUnit 的分层规则也管不到它。
@Configuration
class WorkbenchExecutorConfig {
@Bean("workbenchExecutor")
fun workbenchExecutor(): ThreadPoolTaskExecutor = ThreadPoolTaskExecutor().apply {
corePoolSize = 8
maxPoolSize = 16
queueCapacity = 32 // 有界！无界队列会让 maxPoolSize 永远不生效
setThreadNamePrefix("workbench-")
setRejectedExecutionHandler(ThreadPoolExecutor.CallerRunsPolicy())
// 关键：把 MDC（traceId）和 RequestContext（门店上下文）带到子线程
setTaskDecorator(ContextPropagatingTaskDecorator())
setWaitForTasksToCompleteOnShutdown(true)
setAwaitTerminationSeconds(20) // 配合 09 的优雅停机
initialize()
}
}
四个必须这么写的点：
- 必须是专用池，不能用公共
@Async默认池。聚合任务和别的异步任务共用一个池，一个下游变慢就会把池占满，波及所有异步任务——这正是舱壁模式要防的事（见 05-integration-layer.md）。 - 队列必须有界。
ThreadPoolTaskExecutor的queueCapacity默认是Integer.MAX_VALUE，即无界——后果是任务全部堆进队列，线程数永远不会从 core 涨到 max，然后在某次流量高峰把堆内存吃光。 CallerRunsPolicy：池满时任务退回调用线程（Tomcat 线程）自己执行。这是一种天然的背压——聚合变慢了，但不会丢请求、不会抛RejectedExecutionException。ContextPropagatingTaskDecorator：Micrometer 提供的上下文传播装饰器，把 MDC 里的 traceId、@RequestScope的门店上下文搬到子线程。没有它，子线程里的日志全部断链，且StoreContextHolder直接取不到值——这是同步栈下并行聚合最典型的翻车方式（08-observability.md 附录也点了这一处）。
整体超时预算
// domains/workbench/src/main/kotlin/.../workbench/application/WorkbenchAppService.kt
@Service
class WorkbenchAppService(
@Qualifier("workbenchExecutor") private val executor: Executor,
private val f6ApiClient: F6ApiClient,
private val o2oClient: O2OClient,
private val storeQueryService: StoreQueryService,
) {
private val log = LoggerFactory.getLogger(javaClass)
companion object {
private val TOTAL_BUDGET = Duration.ofSeconds(3) // 整个首页接口的总预算
}
fun loadHomepage(userId: Long, storeId: Long): HomepageResponse {
val deadline = System.nanoTime() + TOTAL_BUDGET.toNanos()
val procurement = supply("procurement") { f6ApiClient.fetchProcurementList(storeId) }
val warranty = supply("warranty") { o2oClient.fetchWarrantySummary(storeId) }
val store = supply("store") { storeQueryService.findStore(storeId) }
return HomepageResponse(
procurement = await(procurement, deadline, "procurement"),
warranty = await(warranty, deadline, "warranty"),
store = await(store, deadline, "store"),
)
}
private fun <T> supply(tile: String, block: () -> T): CompletableFuture<Tile<T>> =
CompletableFuture.supplyAsync({
runCatching(block)
.map { Tile.ok(it) }
.getOrElse { log.warn("tile={} 加载失败，降级", tile, it); Tile.degraded() }
}, executor)
private fun <T> await(future: CompletableFuture<Tile<T>>, deadlineNanos: Long, tile: String): Tile<T> {
val remaining = deadlineNanos - System.nanoTime()
if (remaining <= 0) return Tile.degraded()
return runCatching { future.get(remaining, TimeUnit.NANOSECONDS) }
.getOrElse { log.warn("tile={} 超出总预算，降级", tile); Tile.degraded() }
}
}
总预算不等于各下游超时之和。三个下游各配 2 秒超时，串行最坏 6 秒、并行最坏 2 秒——但这算的是单次调用，叠上 @Retry（3 次）之后单个 tile 最坏可能到 6 秒。所以必须有一个独立于下游配置的接口级总预算（这里 3 秒），到点就把还没回来的 tile 全部降级返回。没有这道闸，首页接口的最坏耗时是由下游配置的乘积决定的，不可控。
局部降级的响应约定
降级必须对客户端可见，不能悄悄返回空数据——客户端要能区分"这块真的没数据"和"这块没拉到"，才能决定是显示空态还是显示"加载失败，点击重试"。
data class Tile<T>(
val data: T?,
val status: TileStatus, // OK / DEGRADED
) {
companion object {
fun <T> ok(data: T) = Tile(data, TileStatus.OK)
fun <T> degraded() = Tile<T>(null, TileStatus.DEGRADED)
}
}
对应的响应体：
{
"code": 0,
"data": {
"procurement": { "status": "OK", "data": { "pendingCount": 12 } },
"warranty": { "status": "DEGRADED", "data": null },
"store": { "status": "OK", "data": { "storeId": 1, "name": "示例门店" } }
},
"traceId": "..."
}
规则：只要主数据（门店上下文）拿到了，首页接口就返回 code: 0，个别 tile 降级不会让整个接口失败。这正是架构图 Flow 3 里"局部降级"的含义——一个外部系统抖动不应该让用户连首页都打不开。
关键规则
- 跨 domain 读走
-contract接口，跨 domain 写/状态联动走领域事件；两者都不允许直接 import 对方的内部类型。 - 事件类型定义在契约模块；监听器一律
@TransactionalEventListener(AFTER_COMMIT)+@Transactional(REQUIRES_NEW)+ 自己兜住异常。 ApplicationEvent只用于"丢了不致命"的副作用；出现需要重试/不能丢的场景，升级到消息中间件 + 事务性发件箱，不要给现有机制打补丁。- 并行聚合必须用专用有界线程池 +
ContextPropagatingTaskDecorator+CallerRunsPolicy。 - 聚合接口必须有独立于下游配置的总超时预算，超时的 tile 降级返回而不是整体失败。
- 降级状态必须在响应里显式表达（
status: DEGRADED），不能用空数据冒充。
待补充
- 各 tile 的具体超时预算分配（需要先有真实的下游耗时数据）。
- 首页聚合结果的本地缓存策略——见 12-concurrency-and-scheduling.md。