mqtt-plus 架构解析五错误处理与 ErrorAction 聚合策略摘要在很多消息框架里错误处理真正困难的地方不是“一个 listener 抛异常怎么办”而是“同一条消息命中多个 listener 时结果不一致怎么办”。mqtt-plus当前的做法是把单个 listener 的失败交给ErrorHandlingStrategy决策再把多个结果交给ErrorActionAggregator聚合。本文会结合真实源码拆解ErrorAction、默认错误策略、严格优先级聚合规则以及这一版实现目前的边界在哪里。项目地址项目地址https://github.com/mqttplus/mqtt-plus配套的示例工程https://github.com/mqttplus/mqtt-plus-examples如果你对这个方向感兴趣欢迎关注、试用也欢迎一起交流 issue 和 PR。如果这篇文章对你有帮助欢迎点赞、收藏也欢迎给项目一个 Star。前面几篇已经把主链路铺开了第 2 篇讲的是一条消息如何走到MqttListener第 3 篇讲的是 payload 为什么拆成入站/出站两条链第 4 篇讲的是MqttMessageInterceptor为什么只做前后观察不做错误决策那错误到底是谁来决定答案在mqtt-plus-core里其实非常集中ErrorHandlingStrategyDefaultErrorHandlingStrategyErrorActionErrorActionAggregator也正因为它足够集中这一篇特别适合把设计边界讲透。一、这篇文章到底想回答什么这一篇只回答三个问题单个 listener 失败之后框架如何把异常转换成统一动作语义同一条消息命中多个 listener结果不一致时如何做最终聚合当前实现为什么说已经有了清晰的错误决策模型但还不是一个完整的端到端重试框架如果只记住一句话那就是mqtt-plus当前把“错误动作语义”抽象清楚了但把“这个动作如何真正反馈到协议层”刻意收得比较克制。二、先看错误处理主链路先看一条消息在路由阶段发生异常时框架内部会走什么路径。先看全局路径Inbound MQTT MessageDefaultMqttMessageRouter.route()for each matched listenerlistener handling blockcollect ErrorActionErrorActionAggregator.aggregate(actions)再把单个 listener 的处理块展开NoYesbeforeHandle()convertPayload()listenerInvoker.invoke()exception thrown?actions.add(ACKNOWLEDGE)errorHandlingStrategy.onError(...)actions.add(ErrorAction)afterHandle()这条链路的关键点只有两个单个 listener 的失败不会直接在路由器里写死成某种行为而是委托给ErrorHandlingStrategy多个 listener 的结果不是边跑边决定而是先收集到actions最后统一交给ErrorActionAggregator也就是说错误处理在 mqtt-plus 里是两层结构第一层单个 listener 的错误决策第二层同一条消息的多 listener 结果聚合这和很多“catch 后直接打印日志”式的实现很不一样它至少先把决策语义显式建模出来了。不过这里要先提前说清一个边界避免后面读到ACKNOWLEDGE、RETRY、DEAD_LETTER这些动作时产生过度联想当前版本已经把错误结果抽象成了明确的动作语义DefaultMqttMessageRouter也确实会在结尾调用errorActionAggregator.aggregate(actions)但route(...)目前仍然是void聚合结果还没有继续反馈到 adapter 或协议层确认逻辑也就是说第 5 篇讨论的重点首先是“动作语义如何被建模和聚合”而不是“这些动作已经完整驱动了端到端协议行为”。三、ErrorAction为什么要先变成统一语义当前ErrorAction非常简单只有 4 个枚举值ACKNOWLEDGERETRYMANUAL_ACKDEAD_LETTER这个设计最重要的意义不是“枚举值够不够多”而是它把错误处理从“异常对象”转换成了“动作语义”。异常对象适合诊断但不适合聚合。因为不同 listener 抛出的异常类型可能完全不同业务最关心的不是异常类名而是“接下来该怎么处理这条消息”只有先把异常映射成统一动作后面才有可能做一致的聚合规则换句话说ErrorAction是一层抽象压缩上游保留异常上下文做诊断下游只关心动作语义做决策设计决策mqtt-plus没有直接拿异常对象去做聚合而是先把失败结果映射成ErrorAction。这样做的重点不是简化异常而是让“多 listener 的最终动作决策”成为可能。四、单个 listener 失败后谁来决定动作这一层由ErrorHandlingStrategy负责。接口非常克制publicinterfaceErrorHandlingStrategy{ErrorActiononError(MqttListenerDefinitiondefinition,MqttContextcontext,Throwableerror);}这里有三个输入特别重要MqttListenerDefinition可以知道是哪个 listener 出错了MqttContext可以拿到 brokerId、topic、payload、headersThrowable可以拿到真实异常这说明它的设计目标很明确决策必须知道是谁失败了决策必须知道消息上下文是什么决策必须知道具体失败原因而默认实现DefaultErrorHandlingStrategy更克制直接返回returnErrorAction.ACKNOWLEDGE;这意味着当前默认策略并不是“失败就重试”而是默认不把单个 listener 失败升级成全局阻断行为。这背后体现的是一个很现实的取舍如果框架默认就倾向RETRY那很多业务异常都会把消息消费链拖进重复处理如果框架默认ACKNOWLEDGE那框架会更偏向“不中断主链路把是否重试交给业务自行覆盖”设计决策默认错误策略返回ACKNOWLEDGE不是因为失败不重要而是因为框架默认选择“保持消费链连续”把更强的重试、人工确认或死信策略留给自定义ErrorHandlingStrategy去决定。五、多个 listener 结果不一致时为什么必须做聚合这才是第 5 篇真正的主问题。一条 MQTT 消息在 mqtt-plus 里可能命中多个 listener。于是就会出现这种情况Listener A 成功结果相当于ACKNOWLEDGEListener B 失败自定义策略返回RETRYListener C 成功结果相当于ACKNOWLEDGE那这条消息最终应该按什么处理跟多数派走选ACKNOWLEDGE只要有一个失败就RETRY交给某个 listener 的优先级mqtt-plus 当前选的是非常明确的一条路把所有 listener 的动作收集起来再按“最严格动作优先”做聚合。Same MQTT MessageListener A - ACKNOWLEDGEListener B - RETRYListener C - ACKNOWLEDGEErrorActionAggregatorFinal Action RETRY这种模型的最大好处是不会因为“多数成功”就把少数失败吞掉能用统一规则覆盖多 listener 场景最终决策保守而明确不依赖业务侧自己拼装推理这也是为什么这里不能简单用多数决。因为在消息处理这种场景里多数决很容易掩盖真正失败的 listener而一旦失败的 listener 承担的是关键业务逻辑多数决反而会制造“看起来成功、实际上不完整”的结果。六、ErrorActionAggregator的聚合规则到底是什么当前实现比很多人想象得更简单。ErrorActionAggregator.aggregate(actions)的实现本质上就是空集合时返回ACKNOWLEDGE非空时按enum ordinal取最大值也就是说当前优先级直接由枚举声明顺序决定ACKNOWLEDGERETRYMANUAL_ACKDEAD_LETTER也可以理解为越靠后动作越“严格”最终聚合结果总是取最严格的那个ACKNOWLEDGERETRYMANUAL_ACKDEAD_LETTERaggregate() returns the strictest action by ordinal从测试也能看出这条规则是被显式验证过的ACKNOWLEDGE RETRY - RETRYMANUAL_ACK DEAD_LETTER - DEAD_LETTER这种实现的优点很明显规则非常稳定聚合逻辑几乎零歧义后续测试也很好写当然这种做法也有一个前提枚举顺序本身就是架构规则。也就是说ErrorAction的声明顺序不是随便排的而是直接决定聚合优先级。这是一种很轻量的实现方式但也要求维护者对枚举顺序非常谨慎。七、为什么不用“多数决”或者“第一个失败说了算”这里其实有三种常见思路多数决第一个失败说了算最严格动作优先mqtt-plus 当前选择第三种是因为它最适合“同一条消息被多个 listener 独立消费”的模型。如果用多数决会出现一个很危险的场景3 个 listener2 个成功1 个关键 listener 失败最终却被判成ACKNOWLEDGE这样表面上系统吞吐更顺但架构语义其实变模糊了因为失败被“票数”稀释了。如果用“第一个失败说了算”问题又会变成结果依赖 listener 执行顺序顺序不同最终动作可能不同这会破坏行为稳定性所以“最严格动作优先”看起来保守但它有两个非常重要的优点不依赖 listener 顺序不会把失败隐藏在多数成功里这正是消息中间件和事件系统常见的一种架构倾向宁可偏保守也不要把不一致结果包装成成功。八、这一版实现的边界在哪里这一节很重要因为如果不把边界讲清楚读者很容易把 mqtt-plus 想象成一个完整的“错误动作执行引擎”。当前真实实现里DefaultMqttMessageRouter在结尾确实会调用errorActionAggregator.aggregate(actions)但它当前的route(...)方法返回值是void聚合出来的最终ErrorAction也没有继续传递给 adapter 或协议层确认逻辑。这意味着当前版本已经有了单 listener 错误决策模型多 listener 结果聚合模型明确的动作语义但还没有完全打通RETRY如何映射到真实重投或重试行为MANUAL_ACK如何映射到协议层确认控制DEAD_LETTER如何映射到真正的死信投递链路所以更准确的说法应该是当前ErrorAction更像一个已经建好的架构接缝而不是一个已经完整闭环的协议执行层。这不是坏事反而说明 mqtt-plus 在这一版里把“决策语义”先定义清楚了但没有急着把所有协议层行为一次性做满。对于一个还在持续演进的框架来说这种节奏其实是合理的。九、小结第 5 篇真正想讲清楚的不是“异常怎么 catch”而是两层设计单个 listener 的失败由ErrorHandlingStrategy决定动作多个 listener 的结果由ErrorActionAggregator做最终聚合而聚合规则也非常明确空集合默认ACKNOWLEDGE非空集合按枚举优先级取最严格动作从架构上看这套设计最大的价值是它把“错误处理”从模糊的异常传播收敛成了可讨论、可扩展、可聚合的动作语义。但同时也要如实看到它当前的边界ErrorAction已经存在聚合规则已经清楚端到端的协议层动作闭环还没有完全打通这也正好为后面的主题留出了空间。下一篇会进入另一条更偏结构层的问题一个应用同时连接多个 broker 时隔离与共享是如何并存的。系列导航本文是mqtt-plus 架构解析系列的第 5/10 篇。#主题链接1总览分层架构与设计哲学链接2消息路由一条 MQTT 消息如何到达你的MqttListener链接3Payload 序列化与反序列化双链设计的取舍链接4拦截器链MqttMessageInterceptor的扩展点设计链接5错误处理ErrorAction聚合策略的设计逻辑本文6多 Broker 管理如何让一个应用同时连接多个 MQTT 服务链接7动态订阅与重连恢复Reconciler的协调机制链接8Spring Boot 自动装配零件是怎么被粘合起来的链接9测试体系MqttTestTemplate与EmbeddedBroker的设计链接10从内部项目到开源框架mqtt-plus 的抽取过程与决策链接上一篇 拦截器链MqttMessageInterceptor的扩展点设计下一篇 多 Broker 管理如何让一个应用同时连接多个 MQTT 服务