Spring Cloud Alibaba：Nacos 服务治理与 Sentinel 流量控制

@Override
public void register(Registration registration) {
    if (StringUtils.isEmpty(registration.getServiceId())) {
        return;
    }
    
    String serviceId = registration.getServiceId();
    String group = nacosDiscoveryProperties.getGroup();
    
    // 构建 Nacos 实例
    Instance instance = getNacosInstanceFromRegistration(registration);
    
    // 注册到 Nacos
    namingService.registerInstance(serviceId, group, instance);
}

public class BeatReactor {
    private final Map<String, BeatInfo> dome2Beat = new ConcurrentHashMap<>();
    
    public void addBeatInfo(String serviceName, BeatInfo beatInfo) {
        // 提交定时任务
        executorService.schedule(new BeatTask(beatInfo), 
            beatInfo.getPeriod(), TimeUnit.MILLISECONDS);
    }
    
    class BeatTask implements Runnable {
        @Override
        public void run() {
            try {
                // 发送 HTTP PUT /nacos/v1/ns/instance/beat
                serverProxy.sendBeat(beatInfo);
                // 重新调度
                executorService.schedule(this, beatInfo.getPeriod(), TimeUnit.MILLISECONDS);
            } catch (Exception e) {
                // 失败重试
            }
        }
    }
}

@Override
public void subscribe(String serviceName, String groupName, EventListener listener) {
    // 1. 立即获取当前服务列表
    List<Instance> instances = namingService.getAllInstances(serviceName, groupName);
    listener.onEvent(new NamingEvent(serviceName, instances));
    
    // 2. 建立 UDP 推送通道
    String key = ServiceInfo.getKey(serviceName, groupName, clusters);
    notifier.registerListener(key, listener);
    
    // 3. UDP 失败时退化为长轮询
    executorService.execute(new NamingRefreshTask(serviceName, groupName, clusters));
}

// NacosNamingService.getAllInstances 内部
public List<Instance> getAllInstances(String serviceName, String groupName) {
    ServiceInfo serviceInfo = serviceInfoHolder.getServiceInfo(key);
    if (serviceInfo == null || !serviceInfo.isValid()) {
        // 本地缓存 miss，向 Server 拉取
        serviceInfo = updateServiceNow(serviceName, groupName, clusters);
    }
    return serviceInfo.getHosts();
}

public class ClientWorker {
    public void addListeners(String dataId, String group, List<? extends Listener> listeners) {
        CacheData cache = addCacheDataIfAbsent(dataId, group);
        cache.setListeners(listeners);
    }
    
    // 长轮询线程
    class LongPollingRunnable implements Runnable {
        @Override
        public void run() {
            // 1. 获取本地缓存的配置 MD5
            List<CacheData> cacheDatas = cacheMap.values();
            
            // 2. 发起长轮询请求
            // timeout = 30000ms（默认）
            List<String> changedGroups = checkUpdateDataIds(cacheDatas, inInitializingCacheList);
            
            // 3. 如果有变更，拉取最新配置
            for (String groupKey : changedGroups) {
                String[] key = GroupKey.parseKey(groupKey);
                String dataId = key[0], group = key[1], tenant = key[2];
                String content = getServerConfig(dataId, group, tenant, 3000L);
                
                // 4. 更新本地缓存并触发监听器
                CacheData cache = cacheMap.get(GroupKey.getKeyTenant(dataId, group, tenant));
                cache.setContent(content);
            }
            
            // 5. 再次调度
            executorService.execute(this);
        }
    }
}

@Target({ ElementType.TYPE, ElementType.METHOD })
@Retention(RetentionPolicy.RUNTIME)
@Scope("refresh")
public @interface RefreshScope {
}

public class RefreshScope implements Scope, DisposableBean {
    private RefreshScopeCache cache;
    
    @Override
    public Object get(String name, ObjectFactory<?> objectFactory) {
        // 从缓存获取 Bean
        BeanWrapper value = cache.get(name);
        if (value == null) {
            // 首次创建
            value = new BeanWrapper(name, objectFactory);
            cache.put(name, value);
        }
        return value.getBean();
    }
}

public synchronized Set<String> refresh() {
    // 1. 从 Nacos 拉取最新配置
    Map<String, Object> newConfig = extractEnvironment();
    
    // 2. 更新 Environment
    addConfigFilesToEnvironment();
    
    // 3. 比较变更的 keys
    Set<String> keys = changes(newConfig, this.context.getEnvironment()).keySet();
    
    // 4. 发布 EnvironmentChangeEvent
    this.context.publishEvent(new EnvironmentChangeEvent(this.context, keys));
    
    // 5. 销毁 @RefreshScope Bean 缓存
    this.scope.refreshAll();
    
    return keys;
}

@RefreshScope
@Service
public class OrderService {
    @Value("${order.timeout:3000}")
    private int timeout;
    
    private final RestTemplate restTemplate = new RestTemplate();  // 非注入
    
    public void createOrder() {
        // 配置刷新时，Bean 被销毁，restTemplate 变成 null
        restTemplate.postForObject(...);
    }
}

spring:
  datasource:
    hikari:
      maximum-pool-size: 20  # 改这个不能简单刷新

public synchronized Set<String> refresh() {  // synchronized！

@Component
@ConfigurationProperties("thread.pool")
@ConstructorBinding
public class ThreadPoolProperties {
    private final int coreSize;
    private final int maxSize;
    
    // getter...
}

@Component
public class ThreadPoolRefresher {
    @Autowired
    private ThreadPoolExecutor executor;
    
    @EventListener
    public void onRefresh(EnvironmentChangeEvent event) {
        if (event.getKeys().stream().anyMatch(k -> k.startsWith("thread.pool"))) {
            // 动态调整线程池参数，而不是重建 Bean
            executor.setCorePoolSize(newCoreSize);
            executor.setMaximumPoolSize(newMaxSize);
        }
    }
}

spring:
  cloud:
    nacos:
      config:
        server-addr: nacos-server:8848
        file-extension: yaml
        # 主配置
        name: order-service
        # 共享配置
        shared-configs:
          - data-id: common.yaml
            group: DEFAULT_GROUP
            refresh: true
        # 扩展配置
        extension-configs:
          - data-id: redis.yaml
            group: DEFAULT_GROUP
            refresh: true

@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, 
                  int count, boolean prioritized, Object... args) throws Throwable {
    try {
        // 继续进入下一个 Slot
        fireEntry(context, resourceWrapper, node, count, prioritized, args);
        
        // 请求通过，记录 pass
        node.increaseThreadNum();
        node.addPassRequest(count);
        
        if (context.getCurEntry().getOriginNode() != null) {
            context.getCurEntry().getOriginNode().increaseThreadNum();
            context.getCurEntry().getOriginNode().addPassRequest(count);
        }
    } catch (PriorityWaitException ex) {
        node.increaseThreadNum();
        // ...
    } catch (BlockException e) {
        // 被限流，记录 block
        node.increaseBlockQps(count);
        throw e;
    } catch (Throwable e) {
        // 业务异常，记录 exception
        node.increaseExceptionQps(count);
        throw e;
    }
}

public class SlidingWindowMetric {
    private final LeapArray<MetricBucket> data;
    
    public SlidingWindowMetric(int sampleCount, int intervalInMs) {
        // sampleCount = 2（默认）
        // intervalInMs = 1000（1秒）
        this.data = new LeapArray<>(sampleCount, intervalInMs);
    }
    
    public long pass() {
        data.currentWindow().value().addPass(1);
    }
}

static boolean passCheck(FlowRule rule, Context context, DefaultNode node, int acquireCount) {
    double curCount = avgUsedTokens(node, rule);
    if (curCount + acquireCount > rule.getCount()) {
        // 超过阈值，触发限流
        return false;
    }
    return true;
}

// 令牌生成间隔从 coldFactor * stableInterval 逐渐降到 stableInterval
long warmUpPeriodInSec = rule.getWarmUpPeriodSec();
double coldFactor = 3;  // 冷启动因子，默认 3

// 计算请求的等待时间
long waitTime = paceController.canPass(node, acquireCount, rule.getCount());
if (waitTime >= 0) {
    // 等待 waitTime 毫秒后放行
    Thread.sleep(waitTime);
    return true;
} else {
    // 排队超时，拒绝
    return false;
}

int curThreadNum = node.curThreadNum();
if (curThreadNum + acquireCount > rule.getCount()) {
    return false;
}

if (avgRT > rule.getCount()) {
    // 平均 RT 超过阈值，进入熔断
    if (cut.compareAndSet(false, true)) {
        // 设置熔断恢复时间窗口
        resetNextRetryTimestamp(rule);
    }
}

double exceptionRatio = (double) exceptionCount / totalCount;
if (exceptionRatio > rule.getCount()) {
    // 异常比例超过阈值，进入熔断
}

if (exceptionCount > rule.getCount()) {
    // 异常数超过阈值，进入熔断
}

@SentinelResource(value = "queryOrder", blockHandler = "queryOrderBlock")
public Order queryOrder(Long userId, Long orderId) {
    return orderDao.findById(orderId);
}

// ParamFlowSlot 核心逻辑
public boolean checkFlow(ResourceWrapper resourceWrapper, ParamFlowRule rule, int count, Object value) {
    // 1. 获取参数值对应的统计节点
    ParameterMetric metric = ParameterMetricStorage.getParamMetric(resourceWrapper);
    CacheMap<Object, AtomicLong> tokenCounter = metric.getRuleTokenCounter(rule);
    CacheMap<Object, AtomicLong> timeCounter = metric.getRuleTimeCounter(rule);
    
    // 2. 检查是否超过阈值
    long currentCount = tokenCounter.get(value).addAndGet(count);
    if (currentCount > rule.getCount()) {
        return false;  // 限流
    }
    return true;
}

@Activate(group = "provider")
public class SentinelDubboProviderFilter extends BaseSentinelDubboFilter {
    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        String resourceName = getResourceName(invoker, invocation, DubboConfig.getDubboProviderPrefix());
        
        Entry interfaceEntry = null;
        Entry methodEntry = null;
        try {
            // 1. 接口级入口
            interfaceEntry = SphU.entry(invoker.getInterface().getName());
            // 2. 方法级入口
            methodEntry = SphU.entry(resourceName);
            
            return invoker.invoke(invocation);
        } catch (BlockException e) {
            // 返回限流响应
            return new AsyncRpcResult(AsyncRpcResult.newDefaultAsyncResult(
                new SentinelRpcException(e), invocation));
        } finally {
            if (methodEntry != null) methodEntry.exit();
            if (interfaceEntry != null) interfaceEntry.exit();
        }
    }
}

public class FlowRuleManager {
    private static volatile Map<String, List<FlowRule>> flowRules = new HashMap<>();
    
    public static void loadRules(List<FlowRule> rules) {
        flowRules = new HashMap<>();
        // ...
    }
}

// 从 Nacos 读取规则
ReadableDataSource<String, List<FlowRule>> flowRuleDataSource = 
    new NacosDataSource<>(remoteAddress, groupId, dataId,
        source -> JSON.parseObject(source, new TypeReference<List<FlowRule>>() {}));
FlowRuleManager.register2Property(flowRuleDataSource.getProperty());

spring:
  profiles:
    active: dev
  cloud:
    nacos:
      config:
        server-addr: nacos-server:8848
        file-extension: yaml
        # 主配置：order-service.yaml
        name: order-service
        # 共享配置
        shared-configs:
          - data-id: common.yaml
            refresh: true
      discovery:
        server-addr: nacos-server:8848
        metadata:
          version: v1
          region: beijing

@Configuration
public class SentinelConfig {
    @PostConstruct
    public void initRules() {
        List<FlowRule> rules = new ArrayList<>();
        
        // 1. QPS 限流规则
        FlowRule qpsRule = new FlowRule();
        qpsRule.setResource("queryOrder");
        qpsRule.setGrade(RuleConstant.FLOW_GRADE_QPS);
        qpsRule.setCount(1000);  // 1000 QPS
        qpsRule.setControlBehavior(RuleConstant.CONTROL_BEHAVIOR_WARM_UP);
        qpsRule.setWarmUpPeriodSec(10);  // 预热 10 秒
        rules.add(qpsRule);
        
        // 2. 并发线程数限流
        FlowRule threadRule = new FlowRule();
        threadRule.setResource("createOrder");
        threadRule.setGrade(RuleConstant.FLOW_GRADE_THREAD);
        threadRule.setCount(50);  // 最多 50 并发线程
        rules.add(threadRule);
        
        // 3. 熔断降级规则
        List<DegradeRule> degradeRules = new ArrayList<>();
        DegradeRule degradeRule = new DegradeRule();
        degradeRule.setResource("payOrder");
        degradeRule.setGrade(RuleConstant.DEGRADE_GRADE_RT);
        degradeRule.setCount(500);  // RT > 500ms
        degradeRule.setTimeWindow(10);  // 熔断 10 秒
        degradeRule.setMinRequestAmount(5);  // 至少 5 个请求才判断
        degradeRules.add(degradeRule);
        
        FlowRuleManager.loadRules(rules);
        DegradeRuleManager.loadRules(degradeRules);
    }
}

@Service
public class OrderService {
    
    @SentinelResource(
        value = "createOrder",
        blockHandler = "createOrderBlock",      // 限流/降级时的处理方法
        fallback = "createOrderFallback"        // 业务异常时的处理方法
    )
    public Order createOrder(CreateOrderRequest request) {
        // 业务逻辑
        return orderDao.insert(request);
    }
    
    // 方法签名必须与原方法一致，最后加 BlockException 参数
    public Order createOrderBlock(CreateOrderRequest request, BlockException ex) {
        log.warn("订单创建被限流: {}", request);
        throw new BizException("系统繁忙，请稍后再试");
    }
    
    // fallback 处理业务异常
    public Order createOrderFallback(CreateOrderRequest request, Throwable ex) {
        log.error("订单创建异常: {}", request, ex);
        return Order.builder().status(OrderStatus.FAILED).build();
    }
}

// Provider 注册时带上版本号
spring:
  cloud:
    nacos:
      discovery:
        metadata:
          version: v2

// Consumer 通过自定义负载均衡选择指定版本
public class VersionLoadBalancer implements ReactorServiceInstanceLoadBalancer {
    @Override
    public Mono<Response<ServiceInstance>> choose(Request request) {
        String targetVersion = getTargetVersionFromRequest(request);
        
        List<ServiceInstance> instances = getInstances();
        List<ServiceInstance> versionInstances = instances.stream()
            .filter(i -> targetVersion.equals(i.getMetadata().get("version")))
            .collect(Collectors.toList());
        
        if (!versionInstances.isEmpty()) {
            return Mono.just(new DefaultResponse(
                versionInstances.get(ThreadLocalRandom.current().nextInt(versionInstances.size()))
            ));
        }
        
        //  fallback 到默认版本
        return Mono.just(new DefaultResponse(
            instances.get(ThreadLocalRandom.current().nextInt(instances.size()))
        ));
    }
}

@ConfigurationProperties("order")
@Data
public class OrderProperties {
    private int timeout = 3000;
    private int retry = 3;
    private boolean enabled = true;
}

// 使用
@Service
public class OrderService {
    @Autowired
    private OrderProperties orderProperties;
    
    public void createOrder() {
        int timeout = orderProperties.getTimeout();
        // ...
    }
}

时间: 0    1    2    3    4    5    6
      |----|----|----|----|----|----|
QPS:  0    100  0    100  0    100

时间: 0   0.5  1   1.5  2   2.5  3
      |--|--|--|--|--|--|
窗口:  W1  W2  W1  W2  W1  W2