SmartLifecycle之RocketMq优雅停机
前言
线上环境发布重启时，经常会导致mq消费失败，不够优雅，之前对RocketMq了解也不太深入，刚好趁这个机会，解决mq的优雅停机问题，顺带着好好撸一撸RocketMq的源码
问题及解决方案
问题现象
线上环境发布重启时，导致一直报MQ消费失败
分析可能的原因
1、线上发布脚本未先下线服务(注册中心未先移除服务配置)
2、脚本非kill -15 导致服务已拉取到的消息直接抛异常
3、系统内部停机不够优雅，在部分bean shutdown后还在消费消息。导致消息处理失败
对于1、2两点，跟运维确认了，发布脚本没问题，那只可能是第三点了
不多说了，撸源码去。
因为我们公司使用的是内部封装后的RocketMq，
关键类信息：
AlicloudRocketMqConsumerHolder：消费者管理器，主要管理mq消息的订阅消费
主要属性以及方法注释：
Java
@Slf4j
public class AlicloudRocketMqConsumerHolder implements ApplicationRunner {
//死信消息处理器
private List<ConsumeFailEnhance> enhances;
//consumer配置信息，用来生成consumer用的
private List<AlicloudRocketMqConsumer> consumerHandlers;
//所有的consumer，用于消费消息和处理消费
@Getter
private List<Consumer> consumers;
//构造函数，1、创建consumer并订阅消息，2构建死信消息处理器
public AlicloudRocketMqConsumerHolder(AlicloudRocketMqProperties rocketMqProperties,
AlicloudRocketMqConsumerProperties consumerProperties,
List<AlicloudRocketMqConsumer> consumerHandlers,
List<ConsumeFailEnhance> enhances,
Tracer tracer) {
//......此处忽略一万行
//分别订阅实现的消息处理器逻辑
for (AlicloudRocketMqConsumer alicloudRocketMqConsumer : gidAlicloudRocketMqConsumer) {
log.info("alicloud rocketmq consumer [{}] gid [{}] topic [{}] tag subExpression [{}]",
alicloudRocketMqConsumer.getClass().getSimpleName(),
alicloudRocketMqConsumer.getGroupId(),
alicloudRocketMqConsumer.getTopic(),
alicloudRocketMqConsumer.getTag());
consumer.subscribe(alicloudRocketMqConsumer.getTopic(),
MessageSelector.byTag(alicloudRocketMqConsumer.getTag()), (message, context) -> {
Tracer.SpanBuilder spanBuilder = tracer.buildSpan(MqTags.CONSUMER)
.withTag(Tags.COMPONENT.getKey(), MqTags.COMPONENT)
.withTag(Tags.SPAN_KIND, MqTags.CONSUMER)
.withTag(MqTags.GID, alicloudRocketMqConsumer.getGroupId())
.withTag(MqTags.TOPIC, message.getTopic())
.withTag(MqTags.TAG, message.getTag())
.withTag(MqTags.KEY, message.getKey())
.withTag(MqTags.ID, message.getMsgID());
SPIConsumerHook spiConsumerHook = new SPIConsumerHook();
spiConsumerHook.onConsume(message, spanBuilder, tracer);
Span span = spanBuilder.start();
try (Scope scope = tracer.activateSpan(span)) {
log.info("alicloud rocketmq consumer MsgID [{}]", message.getMsgID());
alicloudRocketMqConsumer.handle(message, context);
return Action.CommitMessage;
} catch (Throwable throwable) {
SpanUtils.captureException(span, throwable);
if (message.getReconsumeTimes() >= alicloudRocketMqConsumer.getTriggerNotificationTimes()) {
for (ConsumeFailEnhance enhance : enhances) {
//消息 消费 失败，进行增强处理。
String groupId = alicloudRocketMqConsumer.getGroupId();
if (enhance.support(groupId, message)) {
enhance.enhance(groupId, message, throwable);
}
}
}
log.error("alicloud rocketmq consumer fail", throwable);
return Action.ReconsumeLater;
} finally {
span.finish();
}
});
}
consumers.add(consumer);
});
}
}
//..........此处忽略一万行代码
/**
* spring容器启动成功之后，启动consumer
*
* @param applicationArguments应用参数
*/
//启动consumer 进行消息的消费
@Override
public void run(ApplicationArguments applicationArguments) {
long startTime = System.currentTimeMillis();
if (CollectionUtils.isNotEmpty(consumers)) {
log.info("alicloud rocket mq consumer loading！consumer size [{}]", consumers.size());
consumers.forEach(Admin::start);
log.info("alicloud rocket mq consumer启动耗时[{}]", System.currentTimeMillis() - startTime);
}
}
}
可以看到这个叫做消息管理器的类，主要就只提供了两能力，consumer的订阅和启动consumer的start，那么consumer的shutdown呢？只管开启不管结束了？我们只能继续往下看consumer类的实现了：
Java
package com.aliyun.openservices.ons.api;
/**
*消息消费者接口，用来订阅消息
*/
public interface Consumer extends Admin {
/**
*订阅消息
*
* @param topic
*消息主题
* @param subExpression
*订阅过滤表达式字符串，ONS服务器依据此表达式进行过滤。只支持或运算<br>
* eg: "tag1 || tag2 || tag3"<br>
*如果subExpression等于null或者*，则表示全部订阅
* @param listener
*消息回调监听器
*/
void subscribe(final String topic, final String subExpression, final MessageListener listener);
/**
*订阅消息，可以使用SQL表达式对消息进行过滤，请注意，SQL表达式过滤只针对MQ铂金版用户，公网服务暂时不支持。
*
* @param topic
*消息主题
* @param selector
*订阅消息选择器（可空，表示不做过滤），ONS服务器依据此选择器中的表达式进行过滤。
*目前支持两种表达式语法：{@link ExpressionType#TAG}, {@link ExpressionType#SQL92}
*其中，TAG过滤的效果和上面的接口一致
* @param listener
*消息回调监听器
*/
void subscribe(final String topic, final MessageSelector selector, final MessageListener listener);
/**
*取消某个topic订阅
*
* @param topic要取消的主题.
*/
void unsubscribe(final String topic);
}
Consumer接口主要实现了Admin接口：
Java
public interface Admin {
/**
*检查服务是否已经启动.
*
* @return true如果服务已启动; 其它情况返回<code>false</code>
* @see Admin#isClosed()
*/
boolean isStarted();
/**
*检查服务是否已经关闭
*
* @return true如果服务已关闭; 其它情况返回<code>false</code>
* @see Admin#isStarted()
*/
boolean isClosed();
/**
*启动服务
*/
void start();
/**
*更新凭据 (AccessKey/SecretKey/SecurityToken/ONSChannel)
*动态更新访问凭据
* @param credentialProperties STSToken相关的信息
*/
void updateCredential(Properties credentialProperties);
/**
*关闭服务
*/
void shutdown();
}
找到了，shutdown服务。我们来看看它的调用方有没有RocketMQ的销毁bean的钩子
然而并没有找到RocketMQ的销毁bean的钩子，红框内的内容忽略，后面再讲。
所以问题的原因找到了，当服务停机时，并没有先去调用consumer.shutdown()方法停止consumer的消费消息能力，而是直接关闭和销毁spring上下文容器中的bean对象，导致正在处理的消息消费失败。
解决方案
先撸代码再解释
Java
/**
* @Author: dwq
* @Date: 2021/6/11 10:07上午
*/
@Component
public class RocketMQLifecycle implements SmartLifecycle {
@Autowired(required = false)
AlicloudRocketMqConsumerHolder alicloudRocketMqConsumerHolder;
@Override
public void start() {
System.out.println("bean初始化完成");
}
@Override
public void stop() {
System.out.println("bean销毁前执行");
if(alicloudRocketMqConsumerHolder!=null){
alicloudRocketMqConsumerHolder.getConsumers().forEach(Admin::shutdown);
}
}
@Override
public boolean isRunning() {
return true;
}
}
自己创建一个RocketMQLifecycle类实现Spring的SmartLifecycle接口:
它主要有两个功能：
1、在spring容器初始化bean完成后执行start方法
2、在spirng容器关闭前和spring容器bean销毁前执行stop方法
我们今天主要用到了它第2点功能，当停机时，spring容器bean销毁前去执行stop方法，在stop方法中去调用Consumer.shutdown功能进行mq消费优雅停止消费。这样就避免了消息消费失败的问题。
思考：
为啥不用Spring destroy方法去优雅停止消费呢？
因为destroy方法无法保证执行该方法的顺序性，也就是你不能保证mq的消费者优雅停止在其他类销毁之前先执行，从而也就无法解决消费消息失败的问题。
为啥确定SmartLifecycle接口stop方法一定是在destroy方法之前执行的？
下面异常信息是spring容器销毁过程的堆栈信息
忽略一万行代码从
atorg.springframework.context.support.AbstractApplicationContext.close(AbstractApplicationContext.java:975)
这行异常代码开始撸源码：
Java
protected void doClose() {
// Check whether an actual close attempt is necessary...
if (this.active.get() && this.closed.compareAndSet(false, true)) {
if (logger.isDebugEnabled()) {
logger.debug("Closing " + this);
}
LiveBeansView.unregisterApplicationContext(this);
try {
// Publish shutdown event.
publishEvent(new ContextClosedEvent(this));
}
catch (Throwable ex) {
logger.warn("Exception thrown from ApplicationListener handling ContextClosedEvent", ex);
}
// Stop all Lifecycle beans, to avoid delays during individual destruction.
//1、这里就是SmartLifecycle接口所有的实现类的stop方法调用的源头，
//具体细节想看可以自己去撸下源码
if (this.lifecycleProcessor != null) {
try {
this.lifecycleProcessor.onClose();
}
catch (Throwable ex) {
logger.warn("Exception thrown from LifecycleProcessor on context close", ex);
}
}
//2这里是执行所有的destroy方法
// Destroy all cached singletons in the context's BeanFactory.
destroyBeans();
// Close the state of this context itself.
closeBeanFactory();
// Let subclasses do some final clean-up if they wish...
onClose();
// Reset local application listeners to pre-refresh state.
if (this.earlyApplicationListeners != null) {
this.applicationListeners.clear();
this.applicationListeners.addAll(this.earlyApplicationListeners);
}
// Switch to inactive.
this.active.set(false);
}
}
明显可以看出来SmartLifecycle接口stop方法要优先于destroyBeans方法执行
从这里我们也可以看到spring容器销毁的整个过程是咋样的
1、调用SmartLifecycle stop方法停止组件
2、调用destroy方法销毁容器中的bean
3、关闭容器工厂
4、结束
