289 lines
16 KiB
Markdown
289 lines
16 KiB
Markdown
# 11. 跨域协作与聚合
|
||
|
||
## 决策
|
||
|
||
跨 domain 的**读**走契约模块（`-contract`）里的接口，跨 domain 的**写/状态联动**走 Spring `ApplicationEvent` + `@TransactionalEventListener`，暂不引入消息中间件。`bff-orchestration` / `workbench` 的并行聚合用一个专用的有界线程池，带上下文传播和整体超时预算，按 tile 局部降级。
|
||
|
||
这一篇填的是原来整套文档最大的一个空白：[04-security-auth.md](./04-security-auth.md) 里写了"切店 → 通知 webview-ticket 失效"走事件，但事件机制本身从来没有被定义过。
|
||
|
||
## 一、跨域读：契约模块
|
||
|
||
模块结构和约束见 [01-project-structure.md](./01-project-structure.md)，这里只讲使用规则。
|
||
|
||
```kotlin
|
||
// 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,
|
||
)
|
||
```
|
||
|
||
```kotlin
|
||
// 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) }
|
||
}
|
||
```
|
||
|
||
```kotlin
|
||
// domains/workbench/src/main/kotlin/.../workbench/application/WorkbenchAppService.kt
|
||
@Service
|
||
class WorkbenchAppService(
|
||
private val storeQueryService: StoreQueryService, // 注入的是契约里的接口，不是 identity-store 的实现类
|
||
) { ... }
|
||
```
|
||
|
||
### 契约设计的四条规则
|
||
|
||
1. **契约模型是独立的数据结构，不是 Entity 的别名**。`StoreInfo` 只含调用方真正需要的字段。让它跟着 `StoreEntity` 一起长，等于把内部表结构变成了对外承诺，改一个列名就要动三个 domain。
|
||
2. **契约只承诺调用方需要的最小能力**。不要一上来就写 `findAll()`、`update()`——契约里出现的每个方法都是未来的约束。
|
||
3. **契约变更要向后兼容**。加方法、加可空字段没问题；删方法、改语义要先确认所有调用方，跟对客户端的 API 一个待遇。
|
||
4. **契约模块不含实现、不依赖 Spring Web/JPA**（ArchUnit 会检查，见 [10-testing.md](./10-testing.md)）。
|
||
|
||
### 什么时候不该用契约，而应该重新划边界
|
||
|
||
如果 `workbench` 要用 `identity-store` 的契约方法超过五六个，甚至开始要求对方加"给我拼好这个结构"的定制方法，那说明**边界划错了**——这块逻辑本来就该在一边，或者本来就该单独成域。契约模块变厚是个明确的设计告警，不要靠往里加方法来消化它。
|
||
|
||
## 二、跨域写：领域事件
|
||
|
||
### 为什么写操作不走契约接口
|
||
|
||
`workbench` 直接调 `identityStore.doSomething()` 意味着：workbench 要知道 identity-store 内部该做什么，identity-store 的事务边界被外部方法调用拉长，而且以后每多一个关心"切店"的域，就要在切店逻辑里多加一行调用——切店代码变成一个不断膨胀的通知中心。
|
||
|
||
事件反转了这个依赖方向：**发布方不知道谁在听**。切店只管发一个"门店切换了"的事实，谁关心谁自己订阅。
|
||
|
||
### 事件定义放在契约模块
|
||
|
||
```kotlin
|
||
// 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 发布方的内部类型，边界又破了。
|
||
|
||
### 发布：在事务内发布
|
||
|
||
```kotlin
|
||
// 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`
|
||
|
||
```kotlin
|
||
// 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），串行调用意味着总耗时是各下游耗时之和。同步栈下必须显式用线程池做并行。
|
||
|
||
### 专用线程池
|
||
|
||
```kotlin
|
||
// 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()
|
||
}
|
||
}
|
||
```
|
||
|
||
四个必须这么写的点：
|
||
|
||
1. **必须是专用池，不能用公共 `@Async` 默认池**。聚合任务和别的异步任务共用一个池，一个下游变慢就会把池占满，波及所有异步任务——这正是舱壁模式要防的事（见 [05-integration-layer.md](./05-integration-layer.md)）。
|
||
2. **队列必须有界**。`ThreadPoolTaskExecutor` 的 `queueCapacity` 默认是 `Integer.MAX_VALUE`，即无界——后果是任务全部堆进队列，线程数永远不会从 core 涨到 max，然后在某次流量高峰把堆内存吃光。
|
||
3. **`CallerRunsPolicy`**：池满时任务退回调用线程（Tomcat 线程）自己执行。这是一种天然的背压——聚合变慢了，但不会丢请求、不会抛 `RejectedExecutionException`。
|
||
4. **`ContextPropagatingTaskDecorator`**：Micrometer 提供的上下文传播装饰器，把 MDC 里的 traceId、`@RequestScope` 的门店上下文搬到子线程。**没有它，子线程里的日志全部断链，且 `StoreContextHolder` 直接取不到值——这是同步栈下并行聚合最典型的翻车方式**（[08-observability.md](./08-observability.md) 附录也点了这一处）。
|
||
|
||
### 整体超时预算
|
||
|
||
```kotlin
|
||
// 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 全部降级返回。没有这道闸，首页接口的最坏耗时是由下游配置的乘积决定的，不可控。
|
||
|
||
### 局部降级的响应约定
|
||
|
||
降级必须**对客户端可见**，不能悄悄返回空数据——客户端要能区分"这块真的没数据"和"这块没拉到"，才能决定是显示空态还是显示"加载失败，点击重试"。
|
||
|
||
```kotlin
|
||
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)
|
||
}
|
||
}
|
||
```
|
||
|
||
对应的响应体：
|
||
|
||
```json
|
||
{
|
||
"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](./12-concurrency-and-scheduling.md)。
|
||
|
||
## 参考链接
|
||
|
||
- [Spring: Application Events](https://docs.spring.io/spring-framework/reference/core/beans/context-introduction.html#context-functionality-events)
|
||
- [Spring: `@TransactionalEventListener`](https://docs.spring.io/spring-framework/reference/data-access/transaction/event.html)
|
||
- [Micrometer: Context Propagation](https://docs.micrometer.io/context-propagation/reference/)
|
||
- [microservices.io: Transactional Outbox](https://microservices.io/patterns/data/transactional-outbox.html)
|
||
- [Simon Brown: Modular Monoliths](https://www.youtube.com/watch?v=5OjqD-ow8GE)
|