使用Sentinel作为GRPC限流组件，以及使用redis持久化限流规则

墨蓝 2022-12-06 01:13 306阅读 0赞

# 一.使用sentinel作为GRPC的限流组件 #

sentinel官方[https://github.com/alibaba/Sentinel/wiki/%E4%B8%BB%E6%B5%81%E6%A1%86%E6%9E%B6%E7%9A%84%E9%80%82%E9%85%8D\#grpc][https_github.com_alibaba_Sentinel_wiki_E4_B8_BB_E6_B5_81_E6_A1_86_E6_9E_B6_E7_9A_84_E9_80_82_E9_85_8D_grpc]

在官方文档中告诉我们只需要

Server server = ServerBuilder.forPort(port)
         .addService(new MyServiceImpl()) // 添加自己的服务实现
         .intercept(new SentinelGrpcServerInterceptor()) // 在此处注册拦截器
         .build();

这样就可以使用sentinel作为GRPC的限流组件，但是老项目并没有用到或者我们的部分项目不需要sentinel，这样的话，我们需要自适应判断添加SentinelGrpcServerInterceptor。

我重写了阿里的SentinelGrpcServerInterceptor，毕竟Flow control limit exceeded (server side)这句话不是很适合。

class XXXXSentinelGrpcServerInterceptor(fallbackMessage: String) : SentinelGrpcServerInterceptor() {
    
        private val status = Status.UNAVAILABLE.withDescription(
                fallbackMessage)
    
        private val STATUS_RUNTIME_EXCEPTION = StatusRuntimeException(Status.CANCELLED)
    
        override fun <ReqT, RespT> interceptCall(call: ServerCall<ReqT, RespT>, headers: Metadata?, next: ServerCallHandler<ReqT, RespT>): ServerCall.Listener<ReqT>? {
            val fullMethodName = call.methodDescriptor.fullMethodName
            // Remote address: serverCall.getAttributes().get(Grpc.TRANSPORT_ATTR_REMOTE_ADDR);
            var entry: Entry? = null
            return try {
                entry = SphU.asyncEntry(fullMethodName, EntryType.IN)
                val atomicReferenceEntry = AtomicReference<Entry?>(entry)
                // Allow access, forward the call.
                object : ForwardingServerCallListener.SimpleForwardingServerCallListener<ReqT>(
                        next.startCall(
                                object : ForwardingServerCall.SimpleForwardingServerCall<ReqT, RespT>(call) {
                                    override fun close(status: Status, trailers: Metadata) {
                                        val entry = atomicReferenceEntry.get()
                                        if (entry != null) {
                                            // Record the exception metrics.
                                            if (!status.isOk) {
                                                Tracer.traceEntry(status.asRuntimeException(), entry)
                                            }
                                            // entry exit when the call be closed
                                            entry.exit()
                                        }
                                        super.close(status, trailers)
                                    }
                                }, headers)) {
                    /**
                     * If call was canceled, onCancel will be called. and the close will not be called
                     * so the server is encouraged to abort processing to save resources by onCancel
                     * @see ServerCall.Listener.onCancel
                     */
                    override fun onCancel() {
                        val entry = atomicReferenceEntry.get()
                        if (entry != null) {
                            Tracer.traceEntry(STATUS_RUNTIME_EXCEPTION, entry)
                            entry.exit()
                            atomicReferenceEntry.set(null)
                        }
                        super.onCancel()
                    }
                }
            } catch (e: BlockException) {
                call.close(status, Metadata())
                object : ServerCall.Listener<ReqT>() {}
            } catch (e: RuntimeException) {
                // Catch the RuntimeException startCall throws, entry is guaranteed to exit.
                if (entry != null) {
                    Tracer.traceEntry(e, entry)
                    entry.exit()
                }
                throw e
            }
        }
    }

在GRPC注册server时，只需要判断这个bean存在不存在即可

val beanNames = beanFactory.getBeanNamesForType(XXXXSentinelGrpcServerInterceptor::class.java)
                if (beanNames.isNotEmpty()) {
                    logger.info("Register SentinelGrpcServerInterceptor")
                    val XXXXsSentinelGrpcServerInterceptor = beanFactory.getBean(XXXXSentinelGrpcServerInterceptor::class.java)
                    builder.intercept(XXXXSentinelGrpcServerInterceptor)
                }

注：我这里的sentinel组件和GRPC server组件并不在一个服务里。

这时候在sentinel组件里，只需要注册XXXXsSentinelGrpcServerInterceptor这个bean即可。

override fun postProcessBeanDefinitionRegistry(registry: BeanDefinitionRegistry) {
            val beanFactory = registry as ConfigurableListableBeanFactory
    
            val property = Binder.get(environment)
                    .bind("XXXX.sentinel", SentinelProperties::class.java) ?: return
    
                System.setProperty("csp.sentinel.dashboard.server", property.get().serverAddr)
                System.setProperty("project.name", property.get().projectName)
    
            val beanName = "SentinelGrpcServerInterceptor"
                val beanDefinition = BeanDefinitionBuilder.genericBeanDefinition(XXXXSentinelGrpcServerInterceptor::class.java) {
                    val factory = beanFactory.getBean(SentinelTemplateFactory::class.java)
                    factory.createSentinelGrpcServerInterceptor(property.get().fallbackMessage)
                }.beanDefinition
                registry.registerBeanDefinition(beanName, beanDefinition)
    
            val sentinelProperties = "SentinelProperties"
            val sentinelPropertiesDefinition = BeanDefinitionBuilder.genericBeanDefinition(SentinelProperties::class.java) {
                property.get()
            }.beanDefinition
            registry.registerBeanDefinition(sentinelProperties, sentinelPropertiesDefinition)
        }

# 二.使用redis持久化sentinel规则 #

持久化方式，阿里提供了五种持久化方式，分别有file、redis、阿波罗、nacos、zk。我们尽量不引入更多的中间件，所以采用了redis持久化的方式。

阿里留给的持久化接口是InitFunc，我们只需要把这个接口实现了即可。

@Component
    class RedisDataSourceInit : InitFunc {
    
        @Autowired
        private lateinit var sentinelProperties: SentinelProperties
    
        @PostConstruct
        override fun init() {
            val rule = "${sentinelProperties.projectName}:sentinel:rules"
            val flowRule = "$rule:flow-rule"
            val degradeRule = "$rule:degrade-rule"
            val systemRule = "$rule:system-rule"
            val authorityRule = "$rule:authority-rule"
            val hotParamFlowRule = "$rule:param-flow-rule"
    
            // 流控规则
            val parserFlowRule = Converter { source: String -> JSON.parseObject(source, object : TypeReference<List<FlowRule>>() {}) }
            val config: RedisConnectionConfig = RedisConnectionConfig.builder()
                    .withHost(sentinelProperties.redisHost)
                    .withPort(sentinelProperties.redisPort)
                    .build()
            val redisDataSource: ReadableDataSource<String, List<FlowRule>> = RedisDataSource(config, flowRule, flowRule, parserFlowRule)
            FlowRuleManager.register2Property(redisDataSource.property)
    
            val flowRuleWDS: WritableDataSource<List<FlowRule>> = RedisWritableDataSource(sentinelProperties.redisHost, sentinelProperties.redisPort, flowRule)
            WritableDataSourceRegistry.registerFlowDataSource(flowRuleWDS)
    
            // 降级规则
            val parserDegradeRule = Converter { source: String -> JSON.parseObject(source, object : TypeReference<List<DegradeRule>>() {}) }
            val degradeRuleRDS: ReadableDataSource<String, List<DegradeRule>> = RedisDataSource(config, degradeRule, degradeRule, parserDegradeRule)
            DegradeRuleManager.register2Property(degradeRuleRDS.property)
            val degradeRuleWDS: WritableDataSource<List<DegradeRule>> = RedisWritableDataSource(sentinelProperties.redisHost, sentinelProperties.redisPort, degradeRule)
            WritableDataSourceRegistry.registerDegradeDataSource(degradeRuleWDS)
    
            // 系统规则
            val parserSystemRule = Converter { source: String -> JSON.parseObject(source, object : TypeReference<List<SystemRule>>() {}) }
            val systemRuleRDS: ReadableDataSource<String, List<SystemRule>> = RedisDataSource(config, systemRule, systemRule, parserSystemRule)
            SystemRuleManager.register2Property(systemRuleRDS.property)
            val systemRuleWDS: WritableDataSource<List<SystemRule>> = RedisWritableDataSource(sentinelProperties.redisHost, sentinelProperties.redisPort, systemRule)
            WritableDataSourceRegistry.registerSystemDataSource(systemRuleWDS)
    
            // 授权规则
            val parserAuthorityRule = Converter { source: String -> JSON.parseObject(source, object : TypeReference<List<AuthorityRule>>() {}) }
            val authorityRuleRDS: ReadableDataSource<String, List<AuthorityRule>> = RedisDataSource(config, authorityRule, authorityRule, parserAuthorityRule)
            AuthorityRuleManager.register2Property(authorityRuleRDS.property)
            val authorityRuleWDS: WritableDataSource<List<AuthorityRule>> = RedisWritableDataSource(sentinelProperties.redisHost, sentinelProperties.redisPort, authorityRule)
            WritableDataSourceRegistry.registerAuthorityDataSource(authorityRuleWDS)
    
            // 热点参数规则
            val parserParamFlowRule = Converter { source: String -> JSON.parseObject(source, object : TypeReference<List<ParamFlowRule>>() {}) }
            val hotParamFlowRuleRDS: ReadableDataSource<String, List<ParamFlowRule>> = RedisDataSource(config, hotParamFlowRule, hotParamFlowRule, parserParamFlowRule)
            ParamFlowRuleManager.register2Property(hotParamFlowRuleRDS.property)
            val paramFlowRuleWDS: WritableDataSource<List<ParamFlowRule>> = RedisWritableDataSource(sentinelProperties.redisHost, sentinelProperties.redisPort, hotParamFlowRule)
            ModifyParamFlowRulesCommandHandler.setWritableDataSource(paramFlowRuleWDS)
        }
    }

class RedisWritableDataSource<T>(host: String, port: Int, private val key: String) : WritableDataSource<T> {
    
        private val lock: ReentrantLock = ReentrantLock()
    
        private val redisClient: RedisClient = RedisClient.create(RedisURI(host, port, Duration.ofSeconds(2L)))
    
        override fun write(value: T) {
            lock.lock()
            try {
                val toJSONString = JSON.toJSONString(value)
                redisClient.connect().async().set(key, toJSONString)
            } finally {
                lock.unlock()
            }
        }
    
        override fun close() {
            redisClient.shutdown()
        }
    }

[https_github.com_alibaba_Sentinel_wiki_E4_B8_BB_E6_B5_81_E6_A1_86_E6_9E_B6_E7_9A_84_E9_80_82_E9_85_8D_grpc]: https://github.com/alibaba/Sentinel/wiki/%E4%B8%BB%E6%B5%81%E6%A1%86%E6%9E%B6%E7%9A%84%E9%80%82%E9%85%8D#grpc