- 背景
- 引入
2.1 原理 - RabbitMQ实战(实现SPI功能)
3.1 定义抽象父类
3.2 定义功能子类
3.3 实现SPI功能 - RocketMQ和kafka的限流控制
sentinel作为阉割版的限流组件，需要我们进行二次开发才能用于生产环境。当然阿里云也提供了付费版的sentinel（AHAS）可以实现对MQ的削峰填谷。那么开源版如何对MQ实现削峰填谷？？？
1. 背景
在消息队列RocketMQ中，消费者消费消息时，很可能出现因消息发送量突增而消费者来不及处理的情况，导致消费方负载过高，进而导致影响系统稳定性。
在实际场景中，消息的到来具有瞬时性、不规律性，导致系统可能出现空闲资源。利用AHAS应用防护的匀速处理请求的能力，可以把超过消费端处理能力的消息（图中黄色部分）均摊到后面系统空闲时去处理，让系统负载处在一个稳定的水位，同时尽可能地处理更多消息，起到削峰填谷的作用。
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)。
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版消费端削峰填谷
