某互联网公司产品需求在研发与市场间来回踢皮球三个月无进展引入责任链模式后每人负责一环决策周期缩短一半推诿投诉消失
一、三个月的”踢皮球”血泪史
先说点人话。
去年Q3，我们公司有个”重要”项目——帮营销团队做个智能推荐引擎。听起来不高大上吗？简单说就是把用户浏览记录喂给模型，然后给每个人推不同的东西。产品经理老张信心满满地开了个需求评审会，PPT做了八页，讲得唾沫横飞。
结果呢？
研发说”这个需求太模糊，先出技术方案”。
方案出了，市场又说要加一个”实时反馈”功能，研发说”改不了，架构已经定了”。
市场说”那你们先做一个MVP版本”，研发说”MVP也得排期，现在手里有三个项目”。
老张夹在中间，每天在钉钉群里发消息：
@研发小王 这个排期什么时候能确定？ @市场小李 用户调研数据什么时候能给？ @老板 这个需求有没有优先级？
没人回。或者回了但什么都没定。
三个月后，需求文档厚得像本《战争与和平》，但项目进度是：零。
有一次我亲眼看到，老张和产品总监吵起来——老张说”我再不改需求了，你们市场部到底要什么”，总监说”我一开始就要的，是你们研发理解能力不行”。
那周，老张请了两天病假。说是感冒，但我知道不是。
二、转机：一个实习生带来的启发
那个实习生叫小林，后端组的。那天开完会，他凑到我旁边说：
“哥，你们这个需求流转，有没有想过用责任链模式搞一下？”
我当时愣了一下。”责任链？设计模式那玩意儿？能解决现实问题？”
小林笑了笑，打开电脑，写了段伪代码给我看：
// 想象一下你们的需求处理流程
class RequirementChain {
// 每个节点负责一个环节
Handler marketHandler = new MarketHandler(); // 市场：需求收集与确认
Handler productHandler = new ProductHandler(); // 产品：需求文档输出
Handler devHandler = new DevHandler(); // 研发：技术方案与排期
Handler qaHandler = new QAHandler(); // QA：测试计划与验收
// 关键：每个Handler只能处理自己这一环
// 不能越权，不能推诿
marketHandler.setNext(devHandler);
devHandler.setNext(productHandler);
productHandler.setNext(qaHandler);
// 请求从市场发起，链式传递
marketHandler.handle(requirement);
}
我看完愣了一下，然后说：”等等，你的意思是……让每个人只负责自己那一步？”
小林点头：”对啊。现在的问题是，市场说’你们研发做不做’，研发说’你们市场先说清楚要什么’，两个人互踢。但责任链模式里，每个节点只能做自己的事，做完传下去，不能回头踢。”
我盯着那段代码，突然觉得……这可能是个思路。
三、责任链模式：不只是代码，是规则
我们先别急着写代码，先说清楚责任链模式到底是什么。
通俗点说，就是：
有一条链，每个环节只负责一件事。请求从第一个环节开始，每个环节处理完就传给下一个，不能跳，不能回头，不能拒绝。
用你们的场景翻译一下：
| 环节 | 负责人 | 职责 | 输出物 | 决策时限 |
|---|---|---|---|---|
| ① 需求发起 | 市场 | 收集用户反馈，确认需求必要性 | 《需求 brief》 | 2天 |
| ② 需求评审 | 产品 | 输出PRD，确认功能范围 | 《PRD文档》 | 3天 |
| ③ 技术评审 | 研发 | 评估可行性，输出技术方案 | 《技术方案》 | 3天 |
| ④ 排期确认 | 研发 | 排期并确认上线时间 | 《排期表》 | 1天 |
| ⑤ 测试验收 | QA | 测试并出具报告 | 《测试报告》 | 根据用例 |
| ⑥ 上线发布 | 运维 | 部署上线 | 《发布记录》 | 当天 |
关键规则：
- 每个环节必须在时限内完成，超时自动升级给上级
- 每个环节只能做自己的事，不能越权审批别人的环节
- 每个环节的输出物必须是标准格式，下游拿着就能用
- 上游的交付物不合格，下游有权退回，但必须给出明确理由
这不就是把踢皮球变成接力赛吗？
四、代码实现：真正落地的责任链
光有规则不够，得有系统支撑。小林帮我们写了一套简单的责任链处理系统，跑在内部工单平台上。
核心代码大概长这样：
from abc import ABC, abstractmethod
from datetime import datetime, timedelta
from typing import Optional
class Requirement:
"""需求对象"""
def __init__(self, req_id: str, description: str, source: str):
self.req_id = req_id
self.description = description
self.source = source
self.created_at = datetime.now()
self.status = "pending"
self.current_handler = None
self.history = [] # 记录流转历史
def add_history(self, action: str, handler: str):
self.history.append({
"time": datetime.now().isoformat(),
"action": action,
"handler": handler
})
class Handler(ABC):
"""责任链节点基类"""
def __init__(self, name: str, timeout_hours: int = 48):
self.name = name
self.timeout_hours = timeout_hours # 每个节点的决策时限
self.next_handler: Optional[Handler] = None
def set_next(self, handler: 'Handler') -> 'Handler':
"""设置下一个节点"""
self.next_handler = handler
return handler # 支持链式调用
@abstractmethod
def handle(self, requirement: Requirement) -> Requirement:
"""处理需求，返回需求（可能已流转）"""
pass
def try_forward(self, requirement: Requirement) -> Requirement:
"""尝试转发到下一个节点"""
if self.next_handler:
requirement.current_handler = self.next_handler.name
requirement.add_history(f"转发至{self.next_handler.name}", self.name)
return self.next_handler.handle(requirement)
else:
# 已经是最后一个节点
requirement.status = "completed"
requirement.add_history("流程结束", self.name)
return requirement
class MarketHandler(Handler):
"""市场节点：需求发起与确认"""
def handle(self, requirement: Requirement) -> Requirement:
print(f"[{self.name}] 收到需求 #{requirement.req_id}")
requirement.add_history("市场节点处理开始", self.name)
# 模拟：市场需要确认需求的必要性
if not requirement.description:
requirement.add_history("需求描述为空，退回市场补充", self.name)
return requirement # 退回
# 模拟：2天内完成
print(f"[{self.name}] 需求确认完成，准备转发至研发")
requirement.add_history("需求已确认", self.name)
# 转发给下一个节点
return self.try_forward(requirement)
class DevHandler(Handler):
"""研发节点：技术评估与排期"""
def handle(self, requirement: Requirement) -> Requirement:
print(f"[{self.name}] 收到需求 #{requirement.req_id}")
requirement.add_history("研发节点处理开始", self.name)
# 模拟：评估技术方案
if not requirement.description:
requirement.add_history("缺少需求文档，退回产品补充", self.name)
return requirement
# 模拟：3天内完成评估
print(f"[{self.name}] 技术方案已确认，准备转发至产品")
requirement.add_history("技术评估完成", self.name)
return self.try_forward(requirement)
class ProductHandler(Handler):
"""产品节点：PRD输出"""
def handle(self, requirement: Requirement) -> Requirement:
print(f"[{self.name}] 收到需求 #{requirement.req_id}")
requirement.add_history("产品节点处理开始", self.name)
# 模拟：输出PRD
print(f"[{self.name}] PRD文档已输出，准备转发至QA")
requirement.add_history("PRD已输出", self.name)
return self.try_forward(requirement)
class QAHandler(Handler):
"""QA节点：测试与验收"""
def handle(self, requirement: Requirement) -> Requirement:
print(f"[{self.name}] 收到需求 #{requirement.req_id}")
requirement.add_history("QA节点处理开始", self.name)
# 模拟：测试通过
print(f"[{self.name}] 测试通过，流程完成")
requirement.add_history("测试通过", self.name)
return self.try_forward(requirement)
class ChainBuilder:
"""构建责任链"""
@staticmethod
def build() -> Handler:
market = MarketHandler("市场部", timeout_hours=48)
dev = DevHandler("研发部", timeout_hours=72)
product = ProductHandler("产品部", timeout_hours=72)
qa = QAHandler("QA部", timeout_hours=48)
# 串联成链
market.set_next(dev)
dev.set_next(product)
product.set_next(qa)
return market
# ===== 使用示例 =====
if __name__ == "__main__":
# 构建责任链
chain = ChainBuilder.build()
# 创建一个新需求
req = Requirement("REQ-2024-001", "用户推荐引擎优化需求", "市场部")
# 发起流程
print("=" * 50)
print("开始处理需求...")
print("=" * 50)
final_req = chain.handle(req)
# 输出流转记录
print("\n" + "=" * 50)
print("流转记录：")
print("=" * 50)
for record in final_req.history:
print(f"[{record['time']}] {record['handler']}: {record['action']}")
print(f"\n最终状态: {final_req.status}")
运行结果大概是这样的：
==================================================
开始处理需求...
==================================================
[市场部] 收到需求 REQ-2024-001
[市场部] 需求确认完成，准备转发至研发
[研发部] 收到需求 REQ-2024-001
[研发部] 技术方案已确认，准备转发至产品
[产品部] 收到需求 REQ-2024-001
[产品部] PRD文档已输出，准备转发至QA
[QA部] 收到需求 REQ-2024-001
[QA部] 测试通过，流程完成
==================================================
流转记录：
==================================================
[2024-10-15T09:30:00] 市场部: 需求发起
[2024-10-15T09:32:00] 市场部: 需求已确认
[2024-10-15T09:32:00] 市场部: 转发至研发部
[2024-10-15T09:35:00] 研发部: 研发节点处理开始
[2024-10-15T09:38:00] 研发部: 技术评估完成
[2024-10-15T09:38:00] 研发部: 转发至产品部
[2024-10-15T09:40:00] 产品部: 产品节点处理开始
[2024-10-15T09:45:00] 产品部: PRD已输出
[2024-10-15T09:45:00] 产品部: 转发至QA部
[2024-10-15T09:48:00] QA部: QA节点处理开始
[2024-10-15T09:50:00] QA部: 测试通过
[2024-10-15T09:50:00] QA部: 流程结束
最终状态: completed
五、落地之后：真的变了
好，代码说了，模式说了，那真正用起来之后是什么样子？
5.1 老张的故事
还是那个老张。用责任链系统的第一周，他给我发了条消息：
“今天有个需求，市场早上9点提的，下午4点就跑完流程了。以前这种需求三个月都跑不完。”
我问他：”那研发有没有挑刺？”
他说：”挑了，但这次挑得有地方挑——系统里写了’技术方案待确认’，标红了，研发负责人必须在48小时内响应，否则自动升级给CTO。”
原来我们还加了个超时升级机制：
class TimeoutChecker:
"""超时检查器"""
def check(self, requirement: Requirement) -> bool:
"""检查是否超时"""
if requirement.current_handler and requirement.created_at:
elapsed = datetime.now() - requirement.created_at
# 这里简化处理，实际应该按当前节点计算
if elapsed > timedelta(hours=requirement.current_handler.timeout_hours):
# 超时，升级处理
self.escalate(requirement)
return True
return False
def escalate(self, requirement: Requirement):
"""升级到上级"""
print(f"⚠️ 需求 #{requirement.req_id} 超时，升级至上级处理")
requirement.add_history("超时升级", "系统")
# 实际场景中会发送通知给上级
5.2 数据说话
上线责任链系统三个月后，我们拉了数据：
| 指标 | 改革前 | 改革后 | 变化 |
|---|---|---|---|
| 平均决策周期 | 90天 | 45天 | ↓50% |
| 需求一次通过率 | 23% | 67% | ↑191% |
| 跨部门投诉数 | 月均12起 | 月均1起 | ↓92% |
| 需求积压数 | 47个 | 8个 | ↓83% |
| 研发加班时长 | 周均15h | 周均6h | ↓60% |
最让我意外的是最后一个——研发加班少了。以前需求来回扯皮，最后赶工，加班不可避免。现在流程清晰，每个人知道自己该干嘛，干完就完事。
5.3 一个小插曲
当然不是一帆风顺的。
第二个月，市场部的一个新来的同事不愿意用这个系统。她说：”我觉得我可以直接找研发聊，不用走这个流程。”
结果呢？她找了研发小王，小王说：”抱歉，按照责任链规则，没有系统工单我不能接需求，否则出了责任算谁的？”
两人在工位上僵持了五分钟。最后那个同事回去走了流程，系统显示她”跳过市场节点直接给研发提需求”被系统自动拒绝了。
后来她跟我说：”说实话，刚开始挺烦的，但用了一周后发现，系统帮我省去了很多扯皮的麻烦。以前我得追着研发问进度，现在系统里清清楚楚。”
六、责任链模式的精髓：为什么它能解决”踢皮球”
说点更深的。
责任链模式能解决踢皮球问题，核心不是技术，而是规则+透明。
6.1 把”人治”变成”法治”
以前踢皮球的问题是：谁都可以说话，谁都可以否决，但没有一个人必须做决定。
责任链模式下：
- 每个节点有明确的输入和输出
- 每个节点有明确的时限
- 超时自动升级，没人能拖
- 上下游互相可见，没人能藏
6.2 责任与权力对等
以前研发说”需求不明确”，就可以无限期拖。现在不行——系统规定你只有48小时评估，48小时内你不给反馈，系统自动告诉CTO”研发部在拖延”。
权力有了，责任也有了。
6.3 信息透明，信任重建
三个月踢皮球最伤的是什么？不是项目延期，是信任。
市场觉得研发不配合，研发觉得市场不懂技术，产品觉得两边都在甩锅。
责任链系统上线后，所有人的进度都看得见。市场能看到研发在哪个环节卡住了，研发能看到市场的需求文档什么时候交的。
慢慢地，抱怨少了，沟通多了。
七、给想试试这个模式的团队一点建议
如果你们团队也有类似的”踢皮球”问题，想引入责任链模式，我有几点建议：
7.1 不要一上来就写代码
先画流程图，找各方负责人坐下来聊清楚：
- 每个环节的职责边界是什么？
- 每个环节的输出物是什么？
- 每个环节的时限是多少？
这些不搞清楚，写代码也是白写。
7.2 从小范围试点开始
不要一上来就全公司推行。先找一个痛点明显、范围可控的项目试点。我们当初选的就是那个”智能推荐引擎”——因为大家都烦它，而且它不算太大，容易改。
7.3 系统要”温柔地强制”
责任链系统最忌讳”太严”或者”太松”。
- 太严：大家绕过系统走线下，系统形同虚设
- 太松：超时没人管，跟没用一样
我们当时的做法是：前两周系统只记录不惩罚，两周后开始超时提醒，一个月后超时自动升级。
7.4 定期复盘，优化链条
责任链不是设了就完事了。我们每个月会拉数据看：
- 哪个环节经常超时？
- 哪个环节经常被退回？
- 有哪些节点可以合并或拆分？
有一次我们发现”技术评审”环节经常超时，一查是因为研发人手不够。后来调整了排期规则，给这个环节加了个”资源评估”的前置节点，问题就解决了。
八、写在最后：模式只是工具，人才是核心
说了这么多责任链模式，最后想说点题外话。
责任链模式不是魔法。它不能解决所有问题。
我们团队曾经试过另一个项目，用责任链模式跑了一个月，发现还是经常踢皮球。后来才发现——不是模式的问题，是人的问题。有个部门负责人就是不想让项目推进，系统在，但他可以在每个环节”合理”地拖时间。
这时候责任链就失效了。
所以，责任链模式的前提是：各方真的想推进项目，只是之前找不到方法。如果根本动机不对，再好的模式也没用。
但大多数情况下，踢皮球不是因为”不想做”，而是因为不知道谁该做、什么时候该做、做到什么程度算完成。
责任链模式解决的，正是这个问题。
最后分享一个小细节。
项目上线那天，老张在群里发了一条消息：
“三个月了，我终于不用在钉钉上追进度了。”
配图是一张系统截图，显示那个需求已经流转到最后一个节点，状态是”已完成”。
那一刻我觉得，技术真的可以让人工作得更体面一点。