设计模式与消息队列：观察者模式发布消息、命令模式封装消息、责任链处理消息

说实话，消息队列这东西，刚入行时我觉得就是个“中间件”。后来做分布式系统做多了，才明白它背后藏着不少设计模式的智慧。今天咱们就聊聊三个模式在消息队列中的实战配合：观察者模式负责发布、命令模式负责封装、责任链模式负责处理。

我个人习惯把这三者看作一条流水线。消息从产生到消费，每个环节都有对应的模式在起作用。你想想看，如果没有这些模式，代码会变成什么样？大概率是if-else满天飞，耦合得跟麻花似的。

观察者模式：消息的发布机制

观察者模式在消息队列里，说白了就是“发布-订阅”的经典实现。我当年第一次接触ActiveMQ时，就被这种解耦方式震撼到了——生产者只管发，消费者只管收，双方互不知道对方的存在。

核心思想：当一个对象状态发生变化时，自动通知所有依赖它的对象。

在Java中，JDK自带的Observable和Observer接口虽然过时了，但思想还在。现在大家更习惯用Guava的EventBus或者Spring的ApplicationEvent。来看个简单的例子：

// Java示例：观察者模式实现消息发布
public class MessagePublisher {
    private List<MessageListener> listeners = new ArrayList<>();
    
    public void subscribe(MessageListener listener) {
        listeners.add(listener);
    }
    
    public void publish(String topic, String payload) {
        MessageEvent event = new MessageEvent(topic, payload);
        for (MessageListener listener : listeners) {
            listener.onMessage(event);
        }
    }
}

这里有个坑，我曾经踩过——同步通知会导致阻塞。如果某个监听器处理慢了，整个发布流程都会被拖住。所以实际项目中，我建议用线程池异步处理，或者直接上消息中间件。

命令模式：消息的封装艺术

命令模式在消息队列里扮演的角色，就是把“做什么”和“怎么做”分开。每条消息都是一个命令对象，包含了执行所需的所有信息。

我记得有个项目需要支持多种消息类型：订单创建、支付回调、库存更新……如果每种消息都单独写处理逻辑，代码会膨胀得很快。用命令模式封装后，每条消息就是一个Command，统一接口，统一处理。

// C++示例：命令模式封装消息
class MessageCommand {
public:
    virtual ~MessageCommand() = default;
    virtual void execute() = 0;
    virtual std::string getType() = 0;
};

class OrderCreateCommand : public MessageCommand {
private:
    OrderData data;
public:
    explicit OrderCreateCommand(const OrderData& d) : data(d) {}
    void execute() override {
        // 处理订单创建逻辑
    }
    std::string getType() override { return "ORDER_CREATE"; }
};

我的经验：命令对象最好支持序列化。这样消息可以跨网络传输，也方便做日志回放。我在做金融交易系统时，就是靠这个特性实现了消息的可靠投递。

责任链模式：消息的流水线处理

消息到了消费者手里，往往不是一步处理完的。可能需要经过：格式校验 → 业务校验 → 数据持久化 → 发送通知。这一串处理步骤，用责任链模式来组织再合适不过。

我曾经接手过一个遗留系统，消息处理逻辑全写在一个大方法里，足足500行。每次改需求都提心吊胆。后来重构时，我把每个处理步骤拆成独立的Handler，用责任链串起来。效果立竿见影——可读性提升了，测试也方便了。

// Java示例：责任链模式处理消息
public abstract class MessageHandler {
    protected MessageHandler next;
    
    public void setNext(MessageHandler handler) {
        this.next = handler;
    }
    
    public void handle(MessageContext context) {
        if (canHandle(context)) {
            process(context);
        }
        if (next != null) {
            next.handle(context);
        }
    }
    
    protected abstract boolean canHandle(MessageContext context);
    protected abstract void process(MessageContext context);
}

// 具体处理器
class ValidationHandler extends MessageHandler {
    protected boolean canHandle(MessageContext ctx) {
        return ctx.getPhase() == Phase.VALIDATION;
    }
    protected void process(MessageContext ctx) {
        // 校验逻辑
        ctx.setPhase(Phase.BUSINESS_CHECK);
    }
}

三者如何协同工作？

这三个模式不是孤立的，它们在实际系统中是串联使用的。我画了一张图来展示它们的关系：

设计模式在消息队列中的协同工作 观察者模式 消息发布者 命令模式 消息封装器 责任链模式 消息处理器链 发布事件 传递命令 校验处理器 业务处理器 持久化处理器 消息从发布到消费，经历三个模式的协同处理

从图中可以看到，消息的流转路径很清晰：

  1. 观察者模式负责把消息广播出去，通知所有订阅者
  2. 命令模式把消息封装成统一格式，携带执行所需的所有上下文
  3. 责任链模式按顺序处理消息，每个环节只关心自己的职责

实际项目中的避坑指南

我曾经在一个高并发项目中，把这三个模式用在了订单消息处理上。当时遇到几个问题，分享出来供你参考：

注意：责任链中的处理器顺序很重要。我曾经把“发送通知”放在“数据持久化”前面，结果消息还没存库就发出去了，用户收到通知后查不到数据，投诉电话被打爆了。

另外，命令对象的序列化版本号要管理好。我有个同事改了命令类的字段，没更新serialVersionUID，结果线上反序列化失败，消息全部丢失。嗯，这种问题排查起来特别痛苦。

总结一下

这三个模式在消息队列中的配合，说白了就是：观察者模式解决“谁通知谁”的问题，命令模式解决“消息长什么样”的问题，责任链模式解决“消息怎么处理”的问题。

我个人觉得，理解了这个配合关系，你再看任何消息中间件的源码，都会觉得豁然开朗。Kafka的消费者组、RabbitMQ的交换机绑定、RocketMQ的消费链……底层都是这些模式的变体。

最后说一句：设计模式不是教条，是前人踩坑后的经验总结。你在项目中用的时候，要根据实际情况灵活调整。比如责任链不一定要线性执行，也可以支持跳过、重试、甚至回滚。这才是设计模式的正确打开方式。

公众号:蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →