开源版sentinel实现对MQ的削峰填谷（Spring实现SP

2021-09-15 本文已影响0人小胖学编程

背景
引入
2.1 原理
RabbitMQ实战(实现SPI功能)
3.1 定义抽象父类
3.2 定义功能子类
3.3 实现SPI功能
RocketMQ和kafka的限流控制

sentinel作为阉割版的限流组件，需要我们进行二次开发才能用于生产环境。当然阿里云也提供了付费版的sentinel（AHAS）可以实现对MQ的削峰填谷。那么开源版如何对MQ实现削峰填谷？？？

1. 背景

在消息队列RocketMQ中，消费者消费消息时，很可能出现因消息发送量突增而消费者来不及处理的情况，导致消费方负载过高，进而导致影响系统稳定性。

在实际场景中，消息的到来具有瞬时性、不规律性，导致系统可能出现空闲资源。利用AHAS应用防护的匀速处理请求的能力，可以把超过消费端处理能力的消息（图中黄色部分）均摊到后面系统空闲时去处理，让系统负载处在一个稳定的水位，同时尽可能地处理更多消息，起到削峰填谷的作用。

image.png

2. 引入

引入sentinel的依赖：

        <dependency>
            <groupId>com.alibaba.csp</groupId>
            <artifactId>sentinel-core</artifactId>
            <version>1.8.1</version>
        </dependency>

2.1 原理

Sentinel提供了编程式的限流API方法：SphU.entry(String key)。

sentnel的控制台.png

key即资源名。

当消费的速度大于sentinel的控制台配置的qps时，将触发限流，实际上抛出的是BlockException异常，捕获住异常时，可sleep一段时间，然后在将其放入到MQ队列中，实现对MQ消费速度的限流。

3. RabbitMQ实战(实现SPI功能)

3.1 定义抽象父类

目的：寻找Spring容器中RabbitMethodInterceptor抽象类所有的子类，完成功能的增强。

import org.aopalliance.intercept.MethodInterceptor;

/**
 * 抽象类，类似于SPI实现服务的自动发现。Spring会自动寻找该抽象类的所有子类
 *
 * @Date 2021-09-15 22:36
 */
public abstract class RabbitMethodInterceptor implements MethodInterceptor {
}

3.2 定义功能子类

以RabbitMQ为例：定义拦截器


/**
 * Sentinel限流
 */
@Component
@Slf4j
public class SentinelHandlerInterceptor extends RabbitMethodInterceptor implements Serializable {

    @Override
    public Object invoke(MethodInvocation invocation) throws Throwable {
        Object proceed;
        Entry entry = null;
        try {
            Object[] arguments = invocation.getArguments();
            Message message = (Message) arguments[1];
            String consumerQueue = message.getMessageProperties().getConsumerQueue();
            String receivedExchange = message.getMessageProperties().getReceivedExchange();
            //限流的key
            entry = SphU.entry(consumerQueue + "/" + receivedExchange);
            proceed = invocation.proceed();
        } catch (BlockException e) {
            //睡眠一会，防止消息一直进入队列
            Thread.sleep(1000);
            //出现异常即RabbitMQ框架会发出unack，即消息重回队列
            throw new RuntimeException("限流了...");
        } finally {
            if (entry != null) {
                entry.exit();
            }
        }
        return proceed;
    }

}

3.3 实现SPI功能

SPI即可以找到某个接口所有的实现类。此处找到RabbitMethodInterceptor抽象类所有的实现类，来进行切面增强。

注意点：若项目中不存在RabbitMethodInterceptor子类，则启动的时候会抛出异常，故使用ObjectProvider装饰来完成懒加载。

@Slf4j
@EnableRabbit
@Configuration
@EnableConfigurationProperties(value = SealRabbitProperty.class)
public class RabbitConfiguration {


    @Autowired
    private ObjectProvider<List<RabbitMethodInterceptor>> rabbitMethodInterceptorProvider;



    @Bean(name = "sealListenerContainer")
    public SimpleRabbitListenerContainerFactory listenerContainerFactory(CachingConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setConcurrentConsumers(1);
        factory.setMaxConcurrentConsumers(10);
        factory.setPrefetchCount(250);
        /* 设置当rabbitmq收到nack/reject确认信息时的处理方式，设为true，扔回queue头部，设为false，丢弃。 */
        factory.setDefaultRequeueRejected(true);
        //将所有功能都加入到切面中
        List<RabbitMethodInterceptor> rabbitMethodInterceptors = rabbitMethodInterceptorProvider.getIfAvailable();
        if(rabbitMethodInterceptors!=null){
            factory.setAdviceChain(rabbitMethodInterceptors.toArray(new Advice[rabbitMethodInterceptors.size()]));
        }
        //自动确认
        factory.setAcknowledgeMode(AcknowledgeMode.AUTO);
        return factory;
    }
}

后续实现，只需要定义RabbitMethodInterceptor接口实现类，并将其放入到Spring容器中，便可以自动完成监听者功能的增强。

此处增强的是sentinel限流控制功能。

4. RocketMQ和kafka的限流控制

详见文章——AHAS为消息队列Apache RocketMQ版消费端削峰填谷