KiPlant 智慧工厂数字孪生平台——微服务架构搭建与多租户慢SQL治理

-- 原始SQL：查某租户某产线过去30天每日OEE
SELECT DATE(collect_time) as day,
       AVG(availability) as avg_availability,
       AVG(performance) as avg_performance,
       AVG(quality) as avg_quality
FROM device_data
WHERE tenant_id = 'tenant_1024'
  AND production_line_id = 'line_A3'
  AND collect_time BETWEEN '2023-11-01' AND '2023-11-30'
GROUP BY DATE(collect_time)
ORDER BY day;

type: ALL
rows: 52000000
Extra: Using where; Using temporary; Using filesort

ALTER TABLE device_data 
ADD INDEX idx_tenant_line_time (tenant_id, production_line_id, collect_time);

type: range
rows: 86000
Extra: Using index condition; Using temporary; Using filesort

# ShardingSphere配置
rules:
  sharding:
    tables:
      device_data:
        actual-data-nodes: ds_0.device_data_${0..15}
        table-strategy:
          standard:
            sharding-column: tenant_id
            sharding-algorithm-name: tenant-hash-mod
    sharding-algorithms:
      tenant-hash-mod:
        type: HASH_MOD
        props:
          sharding-count: 16

@Intercepts({@Signature(type = Executor.class, method = "query", 
    args = {MappedStatement.class, Object.class, RowBounds.class, ResultHandler.class})})
public class TenantShardingInterceptor implements Interceptor {
​
    // 需要分表的表名集合
    private static final Set<String> SHARDED_TABLES = 
        Set.of("device_data", "work_order", "production_plan");
​
    @Override
    public Object intercept(Invocation invocation) throws Throwable {
        MappedStatement ms = (MappedStatement) invocation.getArgs()[0];
        BoundSql boundSql = ms.getBoundSql(invocation.getArgs()[1]);
        String sql = boundSql.getSql().toLowerCase();
        
        for (String table : SHARDED_TABLES) {
            if (sql.contains(table) && !sql.contains("tenant_id")) {
                throw new IllegalStateException(
                    "Query on sharded table [" + table + "] must include tenant_id. " +
                    "SQL: " + sql.substring(0, Math.min(sql.length(), 200)));
            }
        }
        return invocation.proceed();
    }
}

// 租户级限流规则——从Nacos动态加载
@Component
public class TenantFlowRuleManager {
    
    @PostConstruct
    public void init() {
        // 默认每个租户100 QPS，大客户可单独配置
        FlowRule defaultRule = new FlowRule();
        defaultRule.setResource("api_gateway");
        defaultRule.setGrade(RuleConstant.FLOW_GRADE_QPS);
        defaultRule.setCount(100);
        
        // 按租户ID做热点参数限流
        ParamFlowRule tenantRule = new ParamFlowRule();
        tenantRule.setResource("api_gateway");
        tenantRule.setParamIdx(0);  // 第0个参数是tenantId
        tenantRule.setGrade(RuleConstant.FLOW_GRADE_QPS);
        tenantRule.setCount(100);
        // 大客户例外
        tenantRule.setParamFlowItemList(List.of(
            new ParamFlowItem("tenant_vip_001", 500, String.class.getName())
        ));
        
        ParamFlowRuleManager.loadRules(List.of(tenantRule));
    }
}

@Component
public class TenantDataSourceRouter extends AbstractRoutingDataSource {
​
    @Override
    protected Object determineCurrentLookupKey() {
        String tenantId = TenantContext.getCurrentTenantId();
        // 查租户配置：是否有独立数据源
        TenantConfig config = tenantConfigCache.get(tenantId);
        if (config != null && config.hasDedicatedDs()) {
            return "ds_dedicated_" + tenantId;
        }
        return "ds_shared";  // 默认共享数据源
    }
}

// Gateway过滤器：注入租户上下文
@Component
public class TenantContextFilter implements GlobalFilter, Ordered {
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        String tenantId = extractTenantFromJwt(exchange);
        ServerHttpRequest request = exchange.getRequest().mutate()
            .header("X-Tenant-Id", tenantId)
            .build();
        return chain.filter(exchange.mutate().request(request).build());
    }
}

// 下游服务：Feign拦截器透传
@Component
public class TenantFeignInterceptor implements RequestInterceptor {
    @Override
    public void apply(RequestTemplate template) {
        template.header("X-Tenant-Id", TenantContext.getCurrentTenantId());
    }
}

// MQ消息：Header携带
public void sendDeviceDataEvent(DeviceDataEvent event) {
    Message<DeviceDataEvent> msg = MessageBuilder.withPayload(event)
        .setHeader("tenantId", TenantContext.getCurrentTenantId())
        .build();
    rocketMQTemplate.asyncSend("device-data-topic", msg, callback);
}

# 按租户维度的关键指标
- api_request_duration_seconds{tenant_id, service, endpoint}  # 接口延迟
- api_request_total{tenant_id, service, status_code}           # 请求计数
- db_query_duration_seconds{tenant_id, table}                  # DB查询耗时
- device_data_write_tps{tenant_id}                             # 设备数据写入速率
- tenant_quota_usage_ratio{tenant_id, resource_type}           # 配额使用率

# 服务级别
- jvm_memory_used_bytes{service}
- hikari_connections_active{service, pool}  # 连接池活跃数
- sentinel_flow_block_total{service}        # 限流拦截次数
- rocketmq_consumer_lag{group, topic}       # 消费积压

CREATE TABLE oee_daily_summary (
    id BIGINT PRIMARY KEY,
    tenant_id VARCHAR(64) NOT NULL,
    production_line_id VARCHAR(64),
    stat_date DATE,
    availability DECIMAL(5,4),
    performance DECIMAL(5,4),
    quality_rate DECIMAL(5,4),
    oee DECIMAL(5,4),
    total_output INT,
    defect_count INT,
    planned_run_minutes INT,
    actual_run_minutes INT,
    updated_at DATETIME,
    INDEX idx_tenant_date (tenant_id, stat_date)
);