设计模式——责任链模式实战，优雅处理Kafka消息

一、业务背景

Kafka接收消息，需要A，B，C...多种策略做处理，再通过http请求发送给下游。多种策略混在一起很难维护，通过责任链模式把每种策略的代码收敛到自己的Handler中

二、具体设计

1classDiagram
2Handler <|-- StrategyAHandler
3Handler <|-- StrategyBHandler
4Handler <|-- UploadHandler
5Handler: +void handleUserData(UserContext, HandlerChain)
6Handler: +void handleOrderData(OrderContext, HandlerChain)
7class StrategyAHandler {
8+void handleUserData(UserContext, HandlerChain)
9+void handleOrderData(OrderContext, HandlerChain)
10}
11class StrategyBHandler {
12+void handleUserData(UserContext, HandlerChain)
13+void handleOrderData(OrderContext, HandlerChain)
14}
15class UploadHandler {
16+void handleUserData(UserContext, HandlerChain)
17+void handleOrderData(OrderContext, HandlerChain)
18}
19HandlerChain o--o Handler
20class HandlerChain {
21+List~Handler~ chain
22+void doHandle(AbstractContext context)
23}
24HandlerChain --> AbstractContext
25class AbstractContext {
26+JSONObject param
27+int position
28}
29AbstractContext <|-- UserContext
30AbstractContext <|-- OrderContext
31class UserContext {
32+UserInfo userInfo
33}
34class OrderContext {
35+OrderInfo orderInfo
36}
37

三、代码实现

1.HandlerChain类

1public class HandlerChain {
2
3    List<Handler> chain;
4    
5    public HandlerChain() {
6        chain = new ArrayList<>();
7    }
8    
9    public void doHandle(AbstractContext context) {
10        // 截止条件
11        if (context.getPosition() >= chain.size()) {
12            return;
13        }
14        // 获取当前要执行的handler
15        Handler handler = chain.get(context.getPosition());
16        // 移动到下一个要执行的handler位置
17        context.setPosition(context.getPosition() + 1);
18        // 利用Context类的多态，判断执行的具体方法
19        if (context instanceof UserContext) {
20            handler.handleUserData((UserContext) context, this);
21        } else if (context instanceof OrderContext) {
22            handler.handleOrderData((OrderContext) context, this);
23        }
24    }
25    
26    public void addHandler(Handler handler) {
27        chain.add(hander);
28    }
29
30}
31

2.Handler类

Handler接口：定义需要处理的消息

1public interface Handler {
2    // 处理用户数据消息
3    void handleUserData(UserContext userContext, HandlerChain chain);
4    // 处理订单数据消息
5    void handleOrderData(OrderContext orderContext, HandlerChain chain);
6}
7

Handler实现类：子类结构相同，执行各自的具体逻辑

1@Component
2public class StrategyAHandler implements Handler {
3    // 处理用户数据消息
4    public void handleUserData(UserContext userContext, HandlerChain chain) {
5        // ...
6        // 核心方法，触发流转到下一个handler
7        chain.doHandle(userContext);
8    }
9    // 处理订单数据消息
10    public void handleOrderData(OrderContext orderContext, HandlerChain chain) {
11        ...
12        // 核心方法，触发流转到下一个handler
13        chain.doHandle(orderContext);
14    }
15}
16
17@Component
18public class StrategyBHandler implements Handler {
19 // ...
20}
21
22@Component
23public class UploadHandler implements Handler {
24 // ...
25}
26

3.Context类

Context抽象类：定义公共参数
- param：处理到最后要发http请求，因此构建一个公共的JSONObject类型参数
- position：标记即将执行的handler所在位置

1@Data
2public abstract class AbstractContext {
3    private JSONObject param = new JSONObject();
4    private int position = 0;
5}
6

Context实现类：定义每一种消息独有的参数

1@Data
2public class UserContext extends AbstractContext {
3    private UserInfo userInfo;
4    
5    public UserContext(UserInfo userInfo) {
6        this.userInfo = userInfo;
7    }
8}
9

1@Data
2public class OrderContext extends AbstractContext {
3    private OrderInfo orderInfo;
4    
5    public OrderContext(OrderInfo orderInfo) {
6        this.orderInfo = orderInfo;
7    }
8}
9

4.Handler组装类

作用：接收Kafka消息，把各种策略按任意顺序组装

1@Component
2public class DemoService {
3    @Autowired
4    private StrategyAHandler strategyAHandler;
5    @Autowired
6    private StrategyBHandler strategyBHandler;
7    @Autowired
8    private UploadHandler uploadHandler;
9    
10    // 处理Kafka发来的用户信息
11    public void handleUserData(UserInfo userInfo) {
12        HandlerChain chain = new HandlerChain();
13        chain.addHandler(strategyAHandler);
14        chain.addHandler(strategyBHandler);
15        chain.addHandler(uploadHandler);
16        UserContext userContext = new UserContext(userInfo);
17        chain.doHandle(userContext);
18    }
19    
20    // 处理Kafka发来的订单信息
21    public void handleOrderData(OrderInfo orderInfo) {
22        HandlerChain chain = new HandlerChain();
23        chain.addHandler(strategyBHandler);
24        chain.addHandler(strategyAHandler);
25        chain.addHandler(uploadHandler);
26        OrderContext orderContext = new OrderContext(orderInfo);
27        chain.doHandle(orderContext);
28    }
29}
30

四、总结

Handler实现类交给Spring框架来管理，Context和HandlerChain则是每次调用创建一份
实现效果是可以把某个策略的代码收敛在一个Handler实现类中
如果需要去掉某个策略，直接把该策略从chain中去除即可

《设计模式——责任链模式实战，优雅处理Kafka消息》是转载文章，点击查看原文。