23 KiB
05. 集成层设计(F6 / Mini 域 / 阿里云 OCR)
决策
供应商(F6)、历史 Mini 域、以及阿里云 OCR(车牌识别)的调用统一收口在 integration/f6-adapter / integration/mini-clients / integration/aliyun-ocr 模块(模块位置见 01-project-structure.md),业务 domain 不直接持有 HTTP 客户端;用 Resilience4j 统一管理超时、重试、熔断、并发隔离。
HTTP 客户端用 RestClient(同步),底层是 Apache HttpClient 5 连接池,不用 WebClient/Mono。
为什么是同步 RestClient 而不是 WebClient
WebClient 是 Spring 官方推荐的现代 HTTP 客户端,但它的返回值是 Mono/Flux,把响应式编程模型带进了整个调用链。在我们这套以 Spring MVC(同步 Servlet 栈)为主的应用里,这会造成几个实际问题:
- 上下文会丢。
StoreContextHolder是@RequestScopebean、日志的traceId在 MDC 里,这两样都绑在请求线程上。一旦回调跑在 Reactor 的 event-loop 线程上,两者都拿不到——日志断链、上下文取不到值,而且这类 bug 只在特定时序下出现,很难查。 - 收益用不上。响应式的价值是用少量线程扛住大量并发连接,前提是整条链路都是非阻塞的。我们的链路里有 JPA(阻塞 JDBC),只要还在用 JPA,中间夹一段响应式并不会减少线程占用,只是把复杂度带进来。
- 心智成本。
Mono一旦进入业务代码,测试、异常处理、事务边界都要按另一套规则写,团队要同时掌握两套模型。
RestClient(Spring 6.1+)提供的是和 WebClient 一样的流式 API,但返回值是普通对象、执行是同步的。等到将来整条链路真的要转非阻塞(比如换掉 JPA),再统一切换到 WebClient 不迟——那时是一次有明确收益的整体迁移,而不是现在这样局部引入。
结构约定
platform-integration/
RestClientConfig # 按下游系统构建 RestClient(连接池、超时、拦截器)
IntegrationClientProperties # integration.clients.* 配置绑定
TracePropagationInterceptor # 统一给出站请求加 X-Trace-Id
IntegrationException # 集成层异常基类(继承 platform-web 的 BusinessException)
integration/f6-adapter/
负责:换票、供应商访问上下文准备、超时/重试/熔断/舱壁策略、异常转换为内部标准错误码
integration/mini-clients/
对 O2O/Warranty/Retail Store/ROOS 的只读客户端封装,供 workbench / bff-orchestration 调用
integration/aliyun-ocr/
车牌识别:OSS 临时上传 + RecognizeLicensePlate 调用、AK/SK 持有、按次计费的用量管控
platform-integration 依赖 platform-web(为了复用 BusinessException 和 ErrorCode)是本次允许的少数几个 platform 间依赖之一。
RestClient 配置
// platform-integration/build.gradle
dependencies {
implementation project(':platform:platform-web')
implementation 'org.springframework.boot:spring-boot-starter-web'
implementation 'org.apache.httpcomponents.client5:httpclient5' // 版本由 Boot BOM 管
implementation 'io.github.resilience4j:resilience4j-spring-boot4:2.4.0'
}
// platform-integration/.../RestClientConfig.kt
@Configuration
@EnableConfigurationProperties(IntegrationClientProperties::class)
class RestClientConfig(private val props: IntegrationClientProperties) {
@Bean
fun f6RestClient(builder: RestClient.Builder): RestClient = build(builder, props.f6) { req, resp ->
// 5xx / 连接类问题 → 可重试;4xx → 请求本身有问题,重试没有意义
if (resp.statusCode.is5xxServerError) {
throw F6ServerException("F6 返回 ${resp.statusCode}")
}
throw F6ClientException("F6 拒绝了请求: ${resp.statusCode}")
}
@Bean
fun miniRestClient(builder: RestClient.Builder): RestClient = build(builder, props.mini) { _, resp ->
throw MiniIntegrationException("Mini 域返回 ${resp.statusCode}")
}
private fun build(
builder: RestClient.Builder, // 注入 Boot 提供的 Builder,而不是 RestClient.create():
config: ClientConfig, // 这样 Micrometer 的 observation/W3C traceparent 会自动挂上(见 08)
statusHandler: RestClient.ResponseSpec.ErrorHandler,
): RestClient = builder
.baseUrl(config.baseUrl)
.requestFactory(requestFactory(config))
.requestInterceptor(TracePropagationInterceptor())
.defaultStatusHandler(HttpStatusCode::isError, statusHandler)
.build()
private fun requestFactory(config: ClientConfig): ClientHttpRequestFactory {
val connectionManager = PoolingHttpClientConnectionManagerBuilder.create()
.setMaxConnTotal(config.maxConnections)
.setMaxConnPerRoute(config.maxConnectionsPerRoute) // 必须 ≥ 对应的舱壁并发上限,理由见下
.setDefaultConnectionConfig(
ConnectionConfig.custom()
.setConnectTimeout(Timeout.ofMilliseconds(config.connectTimeoutMs))
.setValidateAfterInactivity(TimeValue.ofSeconds(5)) // 复用前探活,避开对端已断的空闲连接
.build(),
)
.build()
val httpClient = HttpClients.custom()
.setConnectionManager(connectionManager)
.setDefaultRequestConfig(
RequestConfig.custom()
// 从连接池拿连接的等待上限。漏配它是最隐蔽的一个坑:
// 池子被占满时线程会无限期地等在这里,前面所有超时配置全部失效。
.setConnectionRequestTimeout(Timeout.ofMilliseconds(config.connectionRequestTimeoutMs))
.setResponseTimeout(Timeout.ofMilliseconds(config.readTimeoutMs))
.build(),
)
.evictIdleConnections(TimeValue.ofSeconds(30))
.build()
return HttpComponentsClientHttpRequestFactory(httpClient)
}
}
integration:
clients:
f6:
base-url: ${F6_BASE_URL}
connect-timeout-ms: 1000
read-timeout-ms: 2000
connection-request-timeout-ms: 500
max-connections: 60
max-connections-per-route: 30
mini:
base-url: ${MINI_BASE_URL}
connect-timeout-ms: 500
read-timeout-ms: 1000
connection-request-timeout-ms: 300
max-connections: 60
max-connections-per-route: 30
超时由 HTTP 客户端承担,不用 @TimeLimiter。 Resilience4j 的 @TimeLimiter 只能作用于 CompletableFuture 或响应式返回值——同步方法上加了不会报错,但完全不生效,是个非常容易误配的注解。同步栈下真正管用的是上面三个:connectTimeout(建连)、responseTimeout(等响应)、connectionRequestTimeout(等连接池)。
Resilience4j 配置
resilience4j:
# 叠加顺序由这几个 *-aspect-order 属性决定,跟注解写在方法上的先后顺序无关(详见文末附录)。
# 这里显式写死,避免依赖框架默认值——默认值会随版本变,而顺序变了语义就变了。
retry:
retry-aspect-order: 3 # 最外层
instances:
f6-api:
max-attempts: 2
wait-duration: 200ms
exponential-backoff-multiplier: 2
retry-exceptions:
# 同步栈下真实会抛出来的类型:
# - ResourceAccessException 包住了 SocketTimeoutException/ConnectException 等所有 IO 异常,
# 业务代码永远不会直接见到 SocketTimeoutException,写它是无效配置
# - F6ServerException 是我们在 statusHandler 里对 5xx 的转换
- org.springframework.web.client.ResourceAccessException
- com.continental.retailapp.integration.f6.F6ServerException
ignore-exceptions:
# 熔断已打开时抛的异常,重试它毫无意义,只会白白多等一轮 wait-duration
- io.github.resilience4j.circuitbreaker.CallNotPermittedException
- com.continental.retailapp.integration.f6.F6ClientException
circuitbreaker:
circuit-breaker-aspect-order: 2
instances:
f6-api:
sliding-window-type: COUNT_BASED
sliding-window-size: 20
minimum-number-of-calls: 10 # 样本太少时不做判断,避免启动后头几个请求就把熔断打开
failure-rate-threshold: 50
slow-call-duration-threshold: 1500ms
slow-call-rate-threshold: 80 # 慢调用也算故障:只统计失败率的话,"每次都卡满 2 秒但最终成功"永远不会熔断
wait-duration-in-open-state: 10s
permitted-number-of-calls-in-half-open-state: 5
record-exceptions:
- org.springframework.web.client.ResourceAccessException
- com.continental.retailapp.integration.f6.F6ServerException
bulkhead:
bulkhead-aspect-order: 1 # 最内层,贴着真实调用
instances:
f6-api:
max-concurrent-calls: 20
max-wait-duration: 0 # 拿不到名额立刻失败走降级,不排队 —— 排队等于把阻塞换个地方而已
mini-o2o:
max-concurrent-calls: 20
max-wait-duration: 0
F6 客户端示例
// integration/f6-adapter/.../F6ApiClient.kt
@Component
class F6ApiClient(
private val f6RestClient: RestClient, // 来自 platform-integration 的统一封装
) {
@Bulkhead(name = "f6-api") // 默认 SEMAPHORE 类型,不额外起线程
@CircuitBreaker(name = "f6-api", fallbackMethod = "fallbackProcurementList")
@Retry(name = "f6-api")
fun fetchProcurementList(storeId: Long): ProcurementListResponse =
f6RestClient.get()
.uri("/f6/procurement/list?storeId={storeId}", storeId)
.retrieve()
.body(ProcurementListResponse::class.java)!!
// Resilience4j 约定:fallback 方法签名 = 原方法参数 + Throwable,返回类型与原方法一致(同步下就是 T)
fun fallbackProcurementList(storeId: Long, ex: Throwable): ProcurementListResponse {
log.warn("F6 采购列表降级返回,storeId={}, cause={}", storeId, ex.toString())
return ProcurementListResponse.degraded()
}
}
// platform-integration/.../IntegrationException.kt
// 继承 BusinessException(见 06-api-design.md),复用 GlobalExceptionHandler,
// 不再单独写一个 @RestControllerAdvice —— 少一个会和全局处理器抢优先级的地方。
open class IntegrationException(
code: Int,
message: String,
httpStatus: HttpStatus = HttpStatus.BAD_GATEWAY,
) : BusinessException(code, message, httpStatus)
// integration/f6-adapter/.../F6Exceptions.kt
class F6ServerException(message: String) :
IntegrationException(ErrorCode.F6_UNAVAILABLE, "供应商服务暂不可用,请稍后重试")
class F6ClientException(message: String) :
IntegrationException(ErrorCode.F6_BUSINESS_ERROR, "供应商请求被拒绝")
错误码是 Int,落在 3xxxx 集成段(ErrorCode.F6_UNAVAILABLE = 30001),完整分段表见 06-api-design.md——客户端 ../flutter-app/12-error-and-api-contract.md 的 ApiCode 也是数字,两边必须一致。异常里带的原始 message(含 F6 的状态码和响应体)只进日志,不进返回给 APP 的 message,避免把供应商的协议细节泄漏出去。
Mini 域客户端示例(内部系统,策略更宽松)
// integration/mini-clients/.../O2OClient.kt
@Component
class O2OClient(private val miniRestClient: RestClient) {
@Bulkhead(name = "mini-o2o") // 只做超时 + 并发隔离,不加熔断(内部系统,稳定性相对可控)
fun fetchOrderSummary(storeId: Long): OrderSummary =
try {
miniRestClient.get()
.uri("/o2o/orders/summary?storeId={storeId}", storeId)
.retrieve()
.body(OrderSummary::class.java)!!
} catch (ex: Exception) {
log.warn("O2O 订单摘要降级返回,storeId={}", storeId, ex)
OrderSummary.empty() // 局部降级:首页某个 tile 空着,好过整个首页 500
}
}
阿里云 OCR 客户端(车牌识别,按次计费)
车牌识别用阿里云视觉智能开放平台的 RecognizeLicensePlate,客户端不直连、由后端代理,理由与整条链路见 ../flutter-app/07-native-integration.md。它和 F6、Mini 域是同一级别的外部依赖,收口在 integration/aliyun-ocr。
它有两点和前面两个都不一样,配置策略要跟着变:
1. 按次计费,所以不配 @Retry。 F6 重试一次只是多花点时间,OCR 重试一次是多花一次钱,而且识别失败通常是图片本身的问题(模糊、遮挡、角度太偏),重试同一张图大概率还是失败。失败就返回失败,让用户重拍或手工输码——这一条要在代码注释里写清楚,否则后人看到别的 client 都有 @Retry 会以为这里是漏了。
2. 收 ImageURL 不收图片二进制,所以调 OCR 之前必须先把图片落到 OSS。这两步一起构成一次识别,中间任何一步失败对客户端都是"识别失败"。
// integration/aliyun-ocr/.../LicensePlateClient.kt
@Component
class LicensePlateClient(
private val ossClient: OssClient,
private val ocrRestClient: RestClient,
) {
// 只做超时 + 舱壁 + 熔断,不加 @Retry:按次计费,且重试同一张图收益极低
@Bulkhead(name = "aliyun-ocr")
@CircuitBreaker(name = "aliyun-ocr", fallbackMethod = "fallbackRecognize")
fun recognize(image: ByteArray, contentType: String): PlateRecognition {
require(image.size <= 4 * 1024 * 1024) { "图片超过阿里云 4 MB 限制" }
val objectKey = ossClient.putTemp(image, contentType) // 临时对象,见下方生命周期
val resp = ocrRestClient.post()
.uri("/?Action=RecognizeLicensePlate")
.body(mapOf("ImageURL" to ossClient.signedUrl(objectKey)))
.retrieve()
.body(AliyunPlateResponse::class.java)!!
return PlateRecognition(
plateNumber = resp.data.plateNumber,
plateType = resp.data.plateType,
confidence = resp.data.confidence, // 阈值判断交给上层,这里只如实返回
)
}
private fun fallbackRecognize(image: ByteArray, contentType: String, ex: Throwable): PlateRecognition {
log.warn("车牌识别不可用,转手工输码", ex)
throw IntegrationException(ErrorCode.OCR_UNAVAILABLE, "车牌识别暂不可用,请手动输入车牌")
}
}
这里的降级和 Mini 域的"返回空对象"不同:车牌识别没有合理的空值,识别不出来就是识别不出来,只能明确抛错让 APP 切到手工输码。返回一个空字符串会被误当成识别成功。
几条配套约束:
- AK/SK 走 07-config-governance.md 的密钥管理,和数据库口令同级,绝不进代码库、绝不下发到客户端。
- 上传的车辆照片是临时对象:OSS 上设生命周期规则自动过期(识别完就没用了),不要让门店车辆照片无限期堆在桶里——这既是成本问题也是合规问题。
- 要有调用量监控和告警。按次计费的接口一旦被误用(比如客户端做成连续帧调用)账单会很难看,用量指标进 08-observability.md 的看板。
- Region 固定在上海或深圳(该能力只在这两个 Region 提供),配置里写死,不要跟着其它服务的 region 走。
traceId 透传
// platform-integration/.../TracePropagationInterceptor.kt
class TracePropagationInterceptor : ClientHttpRequestInterceptor {
override fun intercept(
request: HttpRequest,
body: ByteArray,
execution: ClientHttpRequestExecution,
): ClientHttpResponse {
MDC.get("traceId")?.let { request.headers.set("X-Trace-Id", it) }
return execution.execute(request, body)
}
}
同步栈下拦截器和业务代码跑在同一个线程上,MDC 一定取得到,业务代码零感知。W3C 标准的 traceparent 头由 Micrometer Tracing 自动加(前提是用注入的 RestClient.Builder 构建,见上面的注释);这里额外发的 X-Trace-Id 是给 F6 这类只认自定义头的外部系统用的,两者并存,规则见 08-observability.md。
关键规则
- F6 是外部供应商域,稳定性不可控,必须配齐超时 + 重试 + 熔断 + 舱壁,且熔断后要有降级返回(
fallbackXxx方法),不能让异常直接穿透到 APP。 - 异常统一转换:F6 / Mini 域的异常和非标准错误,在集成模块内部转换成内部标准错误码,业务 domain 和最终 API 响应都不暴露供应商侧的原始协议细节。
- Mini 域调用相对可控(内部系统),熔断可以不加,但超时和舱壁不能省,避免慢查询拖垮
workbench聚合——对应架构图 Flow 3 "首页失败按 tile 降级"。 - 只对幂等调用配
@Retry。GET 查询可以放心重试;有副作用的调用(下单、扣减)默认不重试,确实需要时接口必须带幂等 key,规则见 12-concurrency-and-scheduling.md。按次计费的调用(阿里云 OCR)同样不重试,理由见上文。 - 降级返回要和业务语义对得上:Mini 域可以返回空对象(首页 tile 空着好过整页 500),但车牌识别不能——识别不出来只能明确抛错让 APP 切手工输码,返回空字符串会被误当成识别成功。
- 不在数据库事务里调外部 HTTP,理由见 03-persistence.md。
- 业务 domain(如
workbench)只依赖integration/*暴露的接口,不自己构造RestClient发请求。 workbench首页把多个下游并行拉起来的写法(线程池、整体超时预算、按 tile 降级)见 11-cross-domain-collaboration.md——那不属于单个客户端的职责。
附录一:超时、重试、熔断、舱壁分别解决什么问题
四者经常被一起提,但作用点不同,配置的时候容易搞混:
-
超时(Timeout):解决"对方一直不回应,我方线程/连接被一直占着"的问题。这是最基础、必须有的一道防线。同步栈下它由 HTTP 客户端提供,不是 Resilience4j 提供的。
-
重试(Retry):解决"这次失败大概率是偶发的(网络抖动、瞬时过载)"的问题。前提是幂等。
-
熔断(Circuit Breaker):解决"对方已经持续故障,继续请求只是在浪费资源、拖慢自己"的问题。统计滑动窗口内的失败率(和慢调用率),超阈值后直接短路走 fallback(
OPEN),过一段时间放几个探测请求(HALF_OPEN)判断是否恢复。 -
舱壁(Bulkhead):解决"一个慢下游把我方所有工作线程吃光"的问题——这一道在同步栈下尤其重要。算笔账:Tomcat 默认 200 个工作线程,F6 读超时 2 秒,如果 F6 全面卡住且没有并发限制,200 个线程会在 2 秒内全部堵在 F6 上,此时这个应用连登录、连查本地数据库的接口都不可用了,一个外部依赖直接把整个服务拖死。配上
max-concurrent-calls: 20之后,最多 20 个线程能进去,剩下 180 个照常干活,超出的请求立刻走降级——用一个功能的降级换整个服务的存活。响应式栈下这道防线的必要性没这么强(event-loop 天生不会被阻塞占死),这也是从
WebClient换到RestClient后必须补上它的原因。舱壁上限和连接池要对齐:
max-connections-per-route必须 ≥max-concurrent-calls,否则真正的瓶颈会变成"抢连接",请求会堵在connectionRequestTimeout上,而不是被舱壁干脆利落地挡掉。
附录二:叠加顺序由配置决定,不是由注解顺序决定
一个常见误解是"注解写在上面的就在外层"——不是的。 Resilience4j 的各个切面都是独立的 Spring AOP Aspect,它们的嵌套顺序由 resilience4j.<type>.<type>-aspect-order 属性(也就是 Spring 的 @Order 语义)决定,跟注解在方法上的书写顺序完全无关。框架的默认顺序是 Retry 在最外层:
Retry ( CircuitBreaker ( RateLimiter ( TimeLimiter ( Bulkhead ( 真实调用 ) ) ) ) )
顺序不同,语义差别很实在:
- Retry 在外(默认,我们采用):每一次重试都会各自经过熔断器,所以一次失败的调用会往熔断器的统计窗口里记 2 笔(
max-attempts: 2)。好处是下游真出问题时熔断器打开得更快;代价是窗口里的样本数会被重试放大,sliding-window-size要按放大后的量来估。这也是为什么必须把CallNotPermittedException加进ignore-exceptions——熔断打开后抛的就是它,不排除掉的话每次请求还要白白多等一轮重试。 - CircuitBreaker 在外:整组重试合起来只在熔断器上记 1 笔,统计更"干净",但下游故障时熔断打开会慢一些。
建议:像上面 yaml 那样把三个 *-aspect-order 显式写出来,不要依赖默认值——默认值可能随版本变化,而这个顺序一变,熔断的触发速度就跟着变了,还很难从现象上察觉。另外,各版本对"order 数值大是更外层还是更内层"的约定容易记反,配完之后用一个必定失败的集成测试实际验证一次叠加顺序(比如断言下游被调用了几次、熔断器记录了几笔),比查文档可靠,写法见 10-testing.md。
待补充
- 具体超时/重试/舱壁参数需要结合 F6 实际 SLA 压测后调整,示例里的数值是起点,不是最终值。
- 熔断后降级返回的数据结构约定(
degraded()的具体字段,以及 APP 侧如何展示"这块数据是降级的")。 - F6 换票的具体协议细节(对接 04-security-auth.md 的切店联动失效)。