移动端网站开发语言,网络维护费,网络营销服务的分类,网页设计在线培训网站有哪些0. 列位#xff0c;响应式布局好麻烦的 …
有意思的#xff0c;chrome devtool 在调试响应式的分辨率的时候#xff0c;比如说在 宽度远远大于 768 的时候#xff0c;按说浏览器也知道大概率是 web端方式打开#xff0c;样式也是如此渲染#xff0c;但一些事件(没有鼠标…0. 列位响应式布局好麻烦的 …
有意思的chrome devtool 在调试响应式的分辨率的时候比如说在 宽度远远大于 768 的时候按说浏览器也知道大概率是 web端方式打开样式也是如此渲染但一些事件(没有鼠标悬停还是维持触控的感觉)却还是移动端上面的。 前面看过了 spring amqp 的官方文档但是它的文档不是按一个个 完整的流程 写的还是老一套 按目录(或者叫 纲目?)拆解 出来的。这就导致我了解了 RetryTemplate, MessageRecoverer, ErrorHandler 的作用但是很难想象他们一起工作的样子。
所以临时起意走一遍源码好久没有做这个事儿了。 本文不会按照这几个组件 加载运行的顺序展开。主打一个 debug 的顺序打开 1. 翻开源码前
debug源码首先需要写一个极简的demo我这边的做法是
配置 关闭 拒绝时重排这个跟本文无关只是补充一下配置开启 消费端 重试代码在消费端抛出一个异常诱发重试现象重试耗尽后消息 转投 死信队列
1.1 最简单的先见名知意
建议看一下 接口定义中 文档注释比较详细
本文默认读者有 RetryTemplate 的经验
MessageRecoverer 消息的恢复器(… 这个确实很抽象作用的位置也深实现非常简单) 如果猜不出来它的作用建议看下他的几个实现类数量很少代码就几行ErrorHandler 错误处理器
1.2 容易纠结的点
因为大概看了一下 运行时堆栈基本跟 MessageRecoverer 不沾边。所以比较纠结—— Retry 拦截器链 跟 ErrorHandler 的关系到底是 一先一后的顺序还是 内外嵌套的结构
解开这个只需要 轻点debug看一下运行时堆栈信息即可
// user method annotated by RabbitListener
onSubscribeTestQueue:93, ExampleQueueClientConfiguration (org.pajamas.example.starter.integration.module.client)invoke0:-1, NativeMethodAccessorImpl (sun.reflect)
invoke:62, NativeMethodAccessorImpl (sun.reflect)
invoke:43, DelegatingMethodAccessorImpl (sun.reflect)
invoke:498, Method (java.lang.reflect)
doInvoke:169, InvocableHandlerMethod (org.springframework.messaging.handler.invocation)
invoke:119, InvocableHandlerMethod (org.springframework.messaging.handler.invocation)
invoke:77, HandlerAdapter (org.springframework.amqp.rabbit.listener.adapter)
invokeHandler:263, MessagingMessageListenerAdapter (org.springframework.amqp.rabbit.listener.adapter)
invokeHandlerAndProcessResult:209, MessagingMessageListenerAdapter (org.springframework.amqp.rabbit.listener.adapter)
org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer#doInvokeListener(org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener, com.rabbitmq.client.Channel, java.lang.Object)
onMessage:148, MessagingMessageListenerAdapter (org.springframework.amqp.rabbit.listener.adapter)
doInvokeListener:1670, AbstractMessageListenerContainer (org.springframework.amqp.rabbit.listener)
actualInvokeListener:1589, AbstractMessageListenerContainer (org.springframework.amqp.rabbit.listener)
invokeListener:-1, 1153028279 (org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer$$Lambda$1982)
invoke0:-1, NativeMethodAccessorImpl (sun.reflect)
invoke:62, NativeMethodAccessorImpl (sun.reflect)
invoke:43, DelegatingMethodAccessorImpl (sun.reflect)
invoke:498, Method (java.lang.reflect)
invokeJoinpointUsingReflection:344, AopUtils (org.springframework.aop.support)
invokeJoinpoint:198, ReflectiveMethodInvocation (org.springframework.aop.framework)
proceed:163, ReflectiveMethodInvocation (org.springframework.aop.framework)
doWithRetry:97, RetryOperationsInterceptor$1 (org.springframework.retry.interceptor)
doExecute:329, RetryTemplate (org.springframework.retry.support)
execute:225, RetryTemplate (org.springframework.retry.support)
invoke:122, RetryOperationsInterceptor (org.springframework.retry.interceptor)
proceed:186, ReflectiveMethodInvocation (org.springframework.aop.framework)// aop拦截器链可以看到只有 Retry 一个advice (上面这一坨都是)
invoke:215, JdkDynamicAopProxy (org.springframework.aop.framework)
invokeListener:-1, $Proxy256 (org.springframework.amqp.rabbit.listener)// 这里是 aop 跟 amqp.listenerContainer 之间的分界线
// 在这里停留看一下 我们代码抛出给 spring.retry 的异常 到这里 怎么是往下走的
// step into ...
invokeListener:1577, AbstractMessageListenerContainer (org.springframework.amqp.rabbit.listener)
doExecuteListener:1568, AbstractMessageListenerContainer (org.springframework.amqp.rabbit.listener)
executeListener:1512, AbstractMessageListenerContainer (org.springframework.amqp.rabbit.listener)
doReceiveAndExecute:994, SimpleMessageListenerContainer (org.springframework.amqp.rabbit.listener)
receiveAndExecute:941, SimpleMessageListenerContainer (org.springframework.amqp.rabbit.listener)
access$1600:85, SimpleMessageListenerContainer (org.springframework.amqp.rabbit.listener)
mainLoop:1319, SimpleMessageListenerContainer$AsyncMessageProcessingConsumer (org.springframework.amqp.rabbit.listener)
run:1225, SimpleMessageListenerContainer$AsyncMessageProcessingConsumer (org.springframework.amqp.rabbit.listener)
run:748, Thread (java.lang)Note: 因为只有一个advice, 即 Retry 的拦截器所以Spring.Retry 与 Spring.Amqp.ErrorHandler 必然是 一先一后 的顺序结构 2. 源码启动
// org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer#executeListener/*** Execute the specified listener, committing or rolling back the transaction afterwards (if necessary).* param channel the Rabbit Channel to operate on* param data the received Rabbit Message* see #invokeListener* see #handleListenerException*/
protected void executeListener(Channel channel, Object data) {if (!isRunning()) {if (logger.isWarnEnabled()) {logger.warn(Rejecting received message(s) because the listener container has been stopped: data);}throw new MessageRejectedWhileStoppingException();}Object sample null;if (this.micrometerHolder ! null) {sample this.micrometerHolder.start();}try {// aop user code doExecuteListener(channel, data);if (sample ! null) {this.micrometerHolder.success(sample, data instanceof Message? ((Message) data).getMessageProperties().getConsumerQueue(): queuesAsListString());}}// ListenerExecutionFailedException(shorten as LEFE) extends AmqpException ( extends RuntimeException )// cause here is AmqpRejectAndDontRequeueException(shorten as ARADRE)catch (RuntimeException ex) {if (sample ! null) {this.micrometerHolder.failure(sample, data instanceof Message? ((Message) data).getMessageProperties().getConsumerQueue(): queuesAsListString(), ex.getClass().getSimpleName());}Message message;if (data instanceof Message) {message (Message) data;}else {message ((ListMessage) data).get(0);}// if message.properties.finalRetryForMessageWithNoId(default false) then rethrowcheckStatefulRetry(ex, message);// ConditionalRejectingErrorHandler.handleError(t)// if is-not-ARADRE exceptionStrategy.isfatal(t)// if // this.errorHandler.discardFatalsWithXDeath(default true) // is-a-LEFE// t has fail message// throw new ImmediateAcknowledgeAmqpException(Fatal and x-death present)// else// throw new AmqpRejectAndDontRequeueException(Error Handler converted exception to fatal);// else // do nothing ...handleListenerException(ex);// step to throw ex;}
}// 最后这一个异常直接跳转到 catch 的地方已经到 main looooooop了
// org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.AsyncMessageProcessingConsumer#mainLoop
// catch ARADRE but do nothing ... (bacuse of it wrapped in LEFE had been handled)
// return main looooooop这里很多注释是我自己写的如果没有看过官方文档也许会对一些名词(比如那些个 异常类型)有些陌生
3. 考虑到读者 也许没有心情 看官方文档
大概 解释一下 为什么要有这几个 异常类型
官网不舍得放图么事我不惜笔墨的 这样看有没有体会到官方文档没有把这俩放在同一个流程中
4. 伪代码启动
现在这个顺序结构已经有了
try {// aop(retry) - invoke listener mothod
} catch {// listenContainer.ErrorHandler ...
}5. 那么好 …
当然源码继续启动看一下几个关键的时刻
5.1 Retry max-attempts exhausted
代码位置可以在控制台的异常日志找到或者 跟踪用户抛出的异常一层层往外抛出时候总会被这里捕获一次的
// org.springframework.retry.support.RetryTemplate#handleRetryExhausted// invoke spring.retry.recover callback// org.springframework.retry.interceptor.RetryOperationsInterceptor.ItemRecovererCallback#recover(retryContext)// org.springframework.amqp.rabbit.config.StatelessRetryOperationsInterceptorFactoryBean#recover(args[], retryContext.latest-throwable)// throw new LEFE(this.messageSupplier.get(), new ARADRE(latest-throwable), message);
/* args: [logger,messageRecoverer, (default RejectAndDontRequeueRecoverer)retryTemplate]
*/MessageRecoverer 做了 new ARADRE(user ex)spring.amqp 的 一个 retry recover-callback 做了 new LEFE(new ARADRE(user ex))
现在对 MessageRecoverer 的理解不难了因为这个案例中我们抛出的异常 不在其配置的strategy 里面不会因为这种失败而触发重排), 这就是 ARADRE 的语义
5.1.1 Locate MessageRecoverer
debug可以看到 这个 MessageRecoverer 不是显式声明出来(step into 被直接跳过了)我们看不到他的声明位置说明是一个lambda 表达式
可以通过所在类 StatelessRetryOperationsInterceptorFactoryBean找进去 protected Object recover(Object[] args, Throwable cause) {// this.messageRecovererMessageRecoverer messageRecoverer getMessageRecoverer();Object arg args[1];if (messageRecoverer null) {this.logger.warn(Message(s) dropped on recovery: arg, cause);}else if (arg instanceof Message) {messageRecoverer.recover((Message) arg, cause);}else if (arg instanceof List messageRecoverer instanceof MessageBatchRecoverer) {((MessageBatchRecoverer) messageRecoverer).recover((ListMessage) arg, cause);}return null;}FactoryBean 是在 Spring.amqp configuration 加载配置的时候用的这是泛泛的说法不难理解
5.1.2 回过头看 configuration 加载配置的地方
// org.springframework.boot.autoconfigure.amqp.RabbitAnnotationDrivenConfiguration#simpleRabbitListenerContainerFactory
// org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer#configure
// org.springframework.boot.autoconfigure.amqp.AbstractRabbitListenerContainerFactoryConfigurer#configure(T, org.springframework.amqp.rabbit.connection.ConnectionFactory, org.springframework.boot.autoconfigure.amqp.RabbitProperties.AmqpContainer)protected void configure(T factory, ConnectionFactory connectionFactory,RabbitProperties.AmqpContainer configuration) {Assert.notNull(factory, Factory must not be null);Assert.notNull(connectionFactory, ConnectionFactory must not be null);Assert.notNull(configuration, Configuration must not be null);factory.setConnectionFactory(connectionFactory);if (this.messageConverter ! null) {factory.setMessageConverter(this.messageConverter);}factory.setAutoStartup(configuration.isAutoStartup());if (configuration.getAcknowledgeMode() ! null) {factory.setAcknowledgeMode(configuration.getAcknowledgeMode());}if (configuration.getPrefetch() ! null) {factory.setPrefetchCount(configuration.getPrefetch());}if (configuration.getDefaultRequeueRejected() ! null) {factory.setDefaultRequeueRejected(configuration.getDefaultRequeueRejected());}if (configuration.getIdleEventInterval() ! null) {factory.setIdleEventInterval(configuration.getIdleEventInterval().toMillis());}factory.setMissingQueuesFatal(configuration.isMissingQueuesFatal());factory.setDeBatchingEnabled(configuration.isDeBatchingEnabled());ListenerRetry retryConfig configuration.getRetry();if (retryConfig.isEnabled()) {RetryInterceptorBuilder?, ? builder (retryConfig.isStateless()) ? RetryInterceptorBuilder.stateless(): RetryInterceptorBuilder.stateful();RetryTemplate retryTemplate new RetryTemplateFactory(this.retryTemplateCustomizers).createRetryTemplate(retryConfig, RabbitRetryTemplateCustomizer.Target.LISTENER);builder.retryOperations(retryTemplate);// set message recovererMessageRecoverer recoverer (this.messageRecoverer ! null) ? this.messageRecoverer: new RejectAndDontRequeueRecoverer();builder.recoverer(recoverer);// step into ...// 这里构造 args[]factory.setAdviceChain(builder.build());}
}// org.springframework.amqp.rabbit.config.RetryInterceptorBuilder#build
// org.springframework.amqp.rabbit.config.RetryInterceptorBuilder.StatelessRetryInterceptorBuilder#build
// org.springframework.amqp.rabbit.config.RetryInterceptorBuilder#applyCommonSettings6. 总结一下 就这样吃饭
