一台故障设备拖垮整个平台：SaaS 系统中"线程池雪崩"的排查与根治

"http-nio-8080-exec-1" TIMED_WAITING
  at java.net.SocketInputStream.read(...)
  at org.apache.http.impl.io.AbstractSessionInputBuffer.read(...)
  at com.kcloud.device.service.VibrationAnalysisClient.analyze(...)
  ...
​
"http-nio-8080-exec-2" TIMED_WAITING
  at java.net.SocketInputStream.read(...)
  ...（同上）
​
... 重复 200 次 ...

客户A的振动分析网关硬件故障，TCP连接假死
    ↓
device-service 调用该网关的HTTP请求一直等待（默认无超时）
    ↓
每个设备诊断请求占用一个Tomcat工作线程
    ↓
客户A有30+台设备持续触发诊断请求，每个请求都卡住不释放
    ↓
Tomcat 200个线程逐步被全部耗尽
    ↓
其他所有客户（B、C、D...）的任何请求都无法获得处理线程
    ↓
全平台不可用

@Configuration
public class RestTemplateConfig {
​
    @Bean
    public RestTemplate vibrationRestTemplate() {
        HttpComponentsClientHttpRequestFactory factory =
            new HttpComponentsClientHttpRequestFactory();
        factory.setConnectTimeout(3000);  // 连接超时 3 秒
        factory.setReadTimeout(5000);     // 读取超时 5 秒
        return new RestTemplate(factory);
    }
}

// Sentinel 熔断降级规则
DegradeRule rule = new DegradeRule();
rule.setResource("vibration-analysis");    // 资源名
rule.setGrade(CircuitBreakerStrategy.ERROR_COUNT.getType()); // 按错误数量
rule.setCount(5);                          // 连续5次错误触发熔断
rule.setTimeWindow(30);                    // 熔断持续30秒
rule.setMinRequestAmount(5);               // 最少5个请求才开始判断
rule.setStatIntervalMs(10000);             // 10秒统计窗口

@SentinelResource(value = "vibration-analysis", fallback = "analysisFallback")
public AnalysisResult analyzeVibration(Long deviceId, VibrationData data) {
    // 正常调用第三方接口
    return vibrationRestTemplate.postForObject(url, data, AnalysisResult.class);
}
​
// 降级处理：记录待重试队列，返回缓存的上一次分析结果
public AnalysisResult analysisFallback(Long deviceId, VibrationData data, Throwable ex) {
    log.warn("设备 {} 振动分析熔断降级，异常: {}", deviceId, ex.getMessage());
​
    // 将任务放入重试队列，等恢复后补偿执行
    retryQueue.add(new RetryTask(deviceId, data));
​
    // 返回该设备上一次的分析结果（从缓存中获取）
    return analysisCache.getLastResult(deviceId)
        .orElse(AnalysisResult.unavailable("诊断服务暂时不可用，将在恢复后自动重试"));
}

@SentinelResource(value = "device-api",
    blockHandler = "tenantLimitHandler")
public Result<?> handleDeviceRequest(
    @SentinelHotParam Long tenantId,  // 按租户ID限流
    DeviceRequest request) {
    return deviceService.process(request);
}
​
public Result<?> tenantLimitHandler(Long tenantId, DeviceRequest request, BlockException ex) {
    log.warn("租户 {} 触发限流", tenantId);
    return Result.fail("请求过于频繁，请稍后重试");
}

ParamFlowRule rule = new ParamFlowRule("device-api")
    .setParamIdx(0)           // 第0个参数（tenantId）
    .setGrade(RuleConstant.FLOW_GRADE_QPS)
    .setCount(200);           // 单个租户最大 200 QPS

设备数据上报 → 校验 → 写入 RocketMQ → 立即返回成功
                                    ↓
                          消费者1：批量写入 TDEngine（时序数据）
                          消费者2：触发振动分析（有熔断保护）
                          消费者3：告警规则判断

// 生产者：设备数据上报接口
@PostMapping("/device/data/report")
public Result<?> reportData(@RequestBody DeviceDataDTO data) {
    // 基础校验
    validateData(data);
    // 异步发送到 MQ，立即返回
    rocketMQTemplate.asyncSend("device-data-topic",
        MessageBuilder.withPayload(data).build(),
        new SendCallback() {
            @Override
            public void onSuccess(SendResult result) {
                log.debug("设备数据投递成功: {}", result.getMsgId());
            }
            @Override
            public void onException(Throwable ex) {
                // 兜底：MQ发送失败时同步写入本地文件，定时任务补偿
                localFileBackup.write(data);
                log.error("MQ投递失败，已写入本地备份", ex);
            }
        });
    return Result.ok();
}

// 消费者：批量消费，提高吞吐
@RocketMQMessageListener(
    topic = "device-data-topic",
    consumerGroup = "device-data-consumer",
    consumeMode = ConsumeMode.CONCURRENTLY,
    consumeThreadNumber = 20
)
public class DeviceDataConsumer implements RocketMQListener<DeviceDataDTO> {

    private final List<DeviceDataDTO> buffer = new ArrayList<>();

    @Override
    public void onMessage(DeviceDataDTO data) {
        buffer.add(data);
        // 攒够100条或超过2秒，批量写入 TDEngine
        if (buffer.size() >= 100) {
            flushToTDEngine();
        }
    }

    @Scheduled(fixedRate = 2000)
    public void scheduledFlush() {
        if (!buffer.isEmpty()) {
            flushToTDEngine();
        }
    }
}

请求进入
   │
   ▼
【第一层：租户级限流 - Sentinel 热点参数】
   │  拦截单个租户的异常流量，保障租户间公平
   ▼
【第二层：异步削峰 - RocketMQ】
   │  高频数据异步化，释放业务线程，平滑流量毛刺
   ▼
【第三层：熔断降级 - Sentinel + 超时控制】
      保护第三方调用，快速失败，避免线程耗尽