一、前言

责任链模式通过构建链式处理结构实现请求的动态传递，其核心优势在于 降低系统耦合度 与 提升扩展性 。主要优点包括：

• 请求发送者与处理者解耦

• 动态扩展能力强

• 职责分离明确

在 Android 开发中，常见的责任链模式应用有：

• ViewGroup 的事件分发责任链

• OkHttp 的拦截器责任链

• Glide 的资源加载责任链

传统责任链模式通常通过对象指针连接处理节点，而 Kotlin Flow 提供了函数式链式调用的新思路。二者对比：

本文重点介绍 Kotlin Flow 实现责任链模式的 4 种实战案例。

二、案例一：基础责任链（数据转换型）

需求场景

实现三个顺序处理步骤：

1. 账号认证处理

2. 输入校验验证

3. 执行结果记录

代码实现

// 1. 定义处理器接口interface ChainHandlerT> { suspend fun process(input: T): Flow}
// 2. 实现三个处理步骤class AuthHandler : ChainHandlerString> { override suspend fun process(input: String): Flow println("AuthHandler 在处理") if (input.startsWith("auth:")) { emit(input.replace("auth:", "") + "[认证完成]") } else { emit(input) } }}
class ValidationHandler : ChainHandlerString> { override suspend fun process(input: String): Flow println("ValidationHandler 在处理") require(input.length >= 3) { "输入长度不足" } emit("校验通过:$input") }}
class LogHandler : ChainHandlerString> { override suspend fun process(input: String): Flow println("处理日志: $input") emit(input) }}
// 3. 构建责任链处理器class ChainProcessorT>(private val handlers: List @OptIn(FlowPreview::class) fun execute(request: T): Flow return handlers.fold(flow { emit(request) }) { acc, handler -> acc.flatMapConcat { handler.process(it) } }.catch { e -> emit("处理异常: ${e.message}" as T) } }}
// 4. 使用示例fun main { runBlocking { val processor = ChainProcessor( listOf( AuthHandler, ValidationHandler, LogHandler ) )
 listOf("auth:test", "short", "normal", "df").forEach { req -> processor.execute(req).flowOn(Dispatchers.IO).collect { println("最终结果: $it") } println("--------") } }}

输出结果

AuthHandler 在处理ValidationHandler 在处理处理日志: 校验通过:test[认证完成]最终结果: 校验通过:test[认证完成]--------AuthHandler 在处理ValidationHandler 在处理处理日志: 校验通过:short最终结果: 校验通过:short--------AuthHandler 在处理ValidationHandler 在处理处理日志: 校验通过:normal最终结果: 校验通过:normal--------AuthHandler 在处理ValidationHandler 在处理最终结果: 处理异常: 输入长度不足--------

设计特点

• 每个处理器都会对数据进行转换

• 使用 fold + flatMapConcat 串联处理链

• 统一的异常处理机制

• 自动背压管理

三、案例二：审批责任链（职责传递型）

需求场景

实现多级审批流程：

• 组长可审批 1 天内请假

• 部门经理可审批 3 天内请假

• 总经理可审批 7 天内请假

代码实现

// 1. 定义数据类data class LeaveRequest( val employee: String, val days: Int, val reason: String, var status: String = "待审批")
// 2. 定义审批处理器接口interface ApprovalHandler { val role: String val maxDays: Int suspend fun process(request: FlowLeaveRequest>): Flow}
// 3. 实现多级审批处理器class TeamLeader : ApprovalHandler { override val role = "组长" override val maxDays = 1 override suspend fun process(request: FlowLeaveRequest>) = request.map { if(it.days "$role 已批准") else it }}
class DepartmentManager : ApprovalHandler { override val role = "部门经理" override val maxDays = 3 override suspend fun process(request: FlowLeaveRequest>) = request.map { if(it.days in 2..maxDays) it.copy(status = "$role 已批准") else it }}
class GeneralManager : ApprovalHandler { override val role = "总经理" override val maxDays = 7 override suspend fun process(request: FlowLeaveRequest>) = request.map { if(it.days in 4..maxDays) it.copy(status = "$role 已批准") else it }}
// 4. 构建审批责任链class ApprovalChain(private val handlers: List @OptIn(FlowPreview::class) suspend fun process(request: LeaveRequest): Flow return handlers.fold(flowOf(request)) { flow, handler -> flow.flatMapConcat { handler.process(flowOf(it)) } }.onEach { if (it.status == "待审批") it.status = "超出权限范围" } }}
// 5. 使用示例fun main { runBlocking { val chain = ApprovalChain( listOf(TeamLeader, DepartmentManager, GeneralManager) )
 listOf( LeaveRequest("张三", 1, "病假"), LeaveRequest("李四", 3, "事假"),  LeaveRequest("王五", 5, "年假"), LeaveRequest("赵六", 10, "婚假") ).forEach { req -> chain.process(req).collect { println("${it.employee}申请${it.days}天: ${it.reason} -> ${it.status}") } println("------------------") } }}

输出结果

张三申请1天: 病假 -> 组长 已批准------------------李四申请3天: 事假 -> 部门经理 已批准 ------------------王五申请5天: 年假 -> 总经理 已批准------------------赵六申请10天: 婚假 -> 超出权限范围------------------

设计特点

• 每个处理器只处理自己职责范围内的请求

• 不满足条件的请求直接传递给下一个处理器

• 所有处理器都会接收到请求但可能不处理

四、案例三：可中断责任链（条件终止型）

需求场景

实现订单处理流程，任何一步失败则终止后续处理：

1. 创建订单

2. 获取支付订单号

3. 生成支付二维码

代码实现

// 1. 定义带中断标记的数据类data class ProcessResultT>( val data: T, val shouldContinue: Boolean = true)
// 2. 定义可中断处理器接口interface BreakableHandlerT> { suspend fun process(input: T): Flow}
// 3. 实现可中断处理器class CreateOrderHandler : BreakableHandlerString> { override suspend fun process(input: String) = flow { println("CreateOrderHandler 在执行") if (input.contains("order")) { emit(ProcessResult("订单号:103202590")) } else { emit(ProcessResult("下单失败", false)) } }}
class GetOrderSnHandler : BreakableHandlerString> { override suspend fun process(input: String) = flow { println("GetOrderSnHandler 在执行") if (input.contains("订单号")) { emit(ProcessResult("支付订单号:1033200")) } else { emit(ProcessResult(input, false)) } }}
class PrePayHandler : BreakableHandlerString> { override suspend fun process(input: String) = flow { println("PrePayHandler 在执行") if (input.contains("支付订单号")) { emit(ProcessResult("支付二维码数据")) } else { emit(ProcessResult(input, false)) } }}
// 4. 构建可中断责任链class BreakableChainT>(private val handlers: List fun execute(input: T): Flow return handlers.fold(flowOf(ProcessResult(input))) { acc, handler -> acc.flatMapConcat { result -> if (!result.shouldContinue) flowOf(result) else handler.process(result.data) } }.map { it.data } }}
// 5. 使用示例fun main { runBlocking { val chain = BreakableChain( listOf(CreateOrderHandler, GetOrderSnHandler, PrePayHandler) )
 listOf("user:order", "invalid").forEach { req -> chain.execute(req).collect { println("最终结果: $it") } println("--------") } }}

输出结果

CreateOrderHandler 在执行GetOrderSnHandler 在执行PrePayHandler 在执行最终结果: 支付二维码数据--------CreateOrderHandler 在执行最终结果: 下单失败--------

设计特点

• 处理器返回是否继续执行的标记

• 任何处理器返回 shouldContinue = false 时终止链

• 后续处理器不会被执行

五、案例四：懒加载责任链（资源优化型）

需求场景

在案例三基础上优化资源使用，只有需要时才创建处理器实例。

代码实现

// 1. 使用工厂函数延迟创建处理器class LazyChainT>(private val handlerFactories: List BreakableHandler fun execute(input: T): Flow return handlerFactories.fold(flowOf(ProcessResult(input))) { acc, factory -> acc.flatMapConcat { result -> if (!result.shouldContinue) flowOf(result) else { println("正在初始化下一级Handler...") factory.process(result.data) } } }.map { it.data } }}
// 2. 添加初始化日志的处理器class CreateOrderHandler : BreakableHandlerString> { init { println("CreateOrderHandler 创建了") }
 override suspend fun process(input: String) = flow { println("CreateOrderHandler 在执行") if (input.contains("order")) { emit(ProcessResult("订单号:103202590")) } else { emit(ProcessResult("下单失败", false)) } }}
// GetOrderSnHandler 和 PrePayHandler 类似添加初始化日志...
// 3. 使用示例fun main { runBlocking { val chain = LazyChain( listOf( { CreateOrderHandler }, { GetOrderSnHandler }, { PrePayHandler } ) )
 listOf("user:order", "invalid").forEach { req -> chain.execute(req).collect { println("最终结果: $it") } println("--------") } }}

输出结果

正在初始化下一级Handler...CreateOrderHandler 创建了CreateOrderHandler 在执行正在初始化下一级Handler...GetOrderSnHandler 创建了GetOrderSnHandler 在执行正在初始化下一级Handler...PrePayHandler 创建了PrePayHandler 在执行最终结果: 支付二维码数据--------正在初始化下一级Handler...CreateOrderHandler 创建了CreateOrderHandler 在执行最终结果: 下单失败--------

设计特点

• 使用工厂函数延迟创建处理器实例

• 只有流程需要时才会创建对应的处理器

• 显著减少资源占用，特别是处理链较长时

• 保持可中断责任链的所有优点

六、总结

本文介绍了四种基于 Kotlin Flow 的责任链模式实现方案：

1. 基础责任链 ：每个处理器都对数据进行转换，适合数据处理流水线场景

2. 审批责任链 ：处理器选择性处理请求，适合职责分离的审批场景

3. 可中断责任链 ：支持条件终止，适合有前置依赖的流程场景

4. 懒加载责任链 ：延迟初始化处理器，优化资源使用

Flow 实现责任链的优势

• 声明式编程 ：代码更简洁易读

• 异步支持 ：天然支持协程异步处理

• 背压管理 ：自动处理生产者-消费者速度差异

• 灵活组合 ：使用操作符轻松组合和变换处理链

• 统一错误处理 ：通过 catch 操作符统一处理异常

适用场景推荐

• 数据处理管道：使用基础责任链模式

• 审批决策流程：使用审批责任链模式

• 有依赖关系的异步操作：使用可中断责任链模式

• 资源敏感场景：使用懒加载责任链模式

Kotlin Flow 为责任链模式带来了更函数式、更声明式的实现方式，特别适合现代异步编程场景，是传统面向对象实现方式的优秀替代方案。