在很早之前，项目出现了一个问题，即有一个消费队列。其中数据处理的过程中出现了OOM，然后导致了RabbitMQ消费者全部挂掉，最终导致消息大量堆积。
在排查这个问题的时候，我深入了解了下RabbitMq源码，最终定位并解决了这个问题。
版本信息：SpringBoot 2.0.4.RELEASE
因为SpringBoot版本不同，RabbitMq源码有一些改动。所以要确定版本号。
1. 问题定位
某个消息出现error级别的异常后，对应的消费者会关闭，且返回unack。消息会回到Mq并转发给下一个消费者去消费。
当然，这个消息在下一个消费者中很大程度上也会出现error异常。故一个error级别的异常，会将整个队列所有的消息者全关闭。
1.1 源码分析
每创建一个消息者，则在线程池中获取一个线程。去执行AsyncMessageProcessingConsumer
创建消费者.png
然后消费者将“死循环”式的监听消费消息。
org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.AsyncMessageProcessingConsumer
消息队列是死循环去消费数据.png
当出现error异常时，如代码所示，会stop掉消费者。
stop消费者.png
2. 解决方案
2.1 方式一：监听事件，重启消费者
在上图可知，当消费者被终止后，会发送Event事件，那么监听事件后重启消费者即可。
@Slf4j
public class ListenerContainerConsumerFailedEventListener implements ApplicationListener<ListenerContainerConsumerFailedEvent> {
/**
* 重启消费者失败的消费者方法
*/
private Consumer<RestartConsumerFailEvent> restartConsumerFailEventConsumer;
/**
* 开启自动重启的回调方法
* true：表示将要重启消费者
*/
private Function<ListenerContainerConsumerFailedEvent, Boolean> failedEventListenerBooleanFunction;
public void setRestartConsumerFailEventConsumer(Consumer<RestartConsumerFailEvent> restartConsumerFailEventConsumer) {
this.restartConsumerFailEventConsumer = restartConsumerFailEventConsumer;
}
public void setFailedEventListenerBooleanFunction(Function<ListenerContainerConsumerFailedEvent, Boolean> failedEventListenerBooleanFunction) {
this.failedEventListenerBooleanFunction = failedEventListenerBooleanFunction;
}
@Override
public void onApplicationEvent(ListenerContainerConsumerFailedEvent event) {
log.error("消费者失败事件发生:{}", event);
//判断是否需要进行重试
Boolean restart = false;
if (failedEventListenerBooleanFunction != null) {
restart = failedEventListenerBooleanFunction.apply(event);
}
if (restart) {
log.error(String.format("Stopping container from aborted consumer. Reason::%s.",
event.getReason()), event.getThrowable());
//获取到容器
SimpleMessageListenerContainer container = (SimpleMessageListenerContainer) event.getSource();
String queueNames = Arrays.toString(container.getQueueNames());
// 重启
try {
restart(container);
log.info("重启队列{}的监听成功！", queueNames);
} catch (Exception e) {
log.error(String.format("重启队列%s的监听失败！", queueNames), e);
//发布事件
if (restartConsumerFailEventConsumer != null) {
RestartConsumerFailEvent restartConsumerFailEvent = new RestartConsumerFailEvent(event.getSource());
restartConsumerFailEvent.setThrowable(e);
restartConsumerFailEventConsumer.accept(restartConsumerFailEvent);
}
}
}
}
/**
* 重启消费者
*
* @param container 容器对象
*/
private void restart(SimpleMessageListenerContainer container) {
try {
Thread.sleep(30000);
} catch (InterruptedException e) {
log.error("", e);
}
Assert.state(!container.isRunning(), String.format("监听容器%s正在运行！", container));
//重启
container.start();
}
}
缺点是：触发error异常的消息依旧没有被消费掉，依旧可能会将重启的消费者给kill掉。
2.2 方式二：捕获Error异常
首先实现拦截器，当出现Error异常后，捕获处理异常，并向外抛出事件通知。
@Slf4j
public class ErrorHandlerInterceptor implements MethodInterceptor, Serializable {
private ApplicationContext applicationContext;
public void setApplicationContext(ApplicationContext applicationContext) {
this.applicationContext = applicationContext;
}
/**
* 当抛出Error异常后，处理方案。
*/
@Override
public Object invoke(MethodInvocation invocation) throws Throwable {
Object proceed = null;
try {
proceed = invocation.proceed();
} catch (Error e) {
log.error("", e);
//遇到Error异常后的处理方案
MessageErrorEvent messageErrorEvent = new MessageErrorEvent(applicationContext, RabbitUtil.getMessage(invocation), e);
applicationContext.publishEvent(messageErrorEvent);
}
return proceed;
}
}
事件信息：
public class MessageErrorEvent extends ApplicationContextEvent {
/**
* 消息对象
*/
private Message message;
/**
* Error异常信息
*/
private Error error;
/**
* Create a new ContextStartedEvent.
*
* @param source the {@code ApplicationContext} that the event is raised for
* (must not be {@code null})
*/
public MessageErrorEvent(ApplicationContext source) {
super(source);
}
public MessageErrorEvent(ApplicationContext source, Message message, Error error) {
super(source);
this.message = message;
this.error = error;
}
public Message getMessage() {
return message;
}
public void setMessage(Message message) {
this.message = message;
}
public Error getError() {
return error;
}
public void setError(Error error) {
this.error = error;
}
}
关键点：将拦截器设置到SimpleRabbitListenerContainerFactory中：
@EnableRabbit
@Configuration
public class RabbitConfiguration {
@Autowired
private ObjectProvider<ErrorHandlerInterceptor> errorHandlerInterceptorObjectProvider;
@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<Advice> adviceList = new ArrayList<>();
ErrorHandlerInterceptor errorHandlerInterceptor = errorHandlerInterceptorObjectProvider.getIfAvailable();
if (errorHandlerInterceptor != null) {
adviceList.add(errorHandlerInterceptor);
}
//加入拦截器配置
List<MessageInterceptor> messageInterceptors = messageInterceptorsObjectProvider.getIfAvailable();
if (!CollectionUtils.isEmpty(messageInterceptors)) {
for (MessageInterceptor messageInterceptor : messageInterceptors) {
if (messageInterceptor != null) {
adviceList.add(messageInterceptor);
}
}
}
factory.setAdviceChain(adviceList.toArray(new Advice[adviceList.size()]));
//自动确认
factory.setAcknowledgeMode(AcknowledgeMode.AUTO);
return factory;
}
}
由此，可以将通过责任链的方式，进行AOP处理，使得消费者不对外抛出Error级别的异常。
网友评论