基于Kafka实现消息中间件：观察者+责任链+命令+工厂

说实话，消息中间件这块，我早年踩过不少坑。那时候公司自研了一套消息系统，结果生产环境频繁丢消息，运维同学半夜打电话把我叫醒……后来我痛定思痛，决定用Kafka重构。但光用Kafka还不够，怎么把业务逻辑和消息处理解耦？怎么让代码好扩展？这时候，设计模式就派上用场了。

今天这个实战案例，我会把四个模式揉在一起：观察者模式负责消息分发，责任链模式处理消息过滤和转换，命令模式封装消息动作，工厂模式统一创建对象。你想想看，这就像一条流水线——消息进来，经过一道道工序，最后被消费掉。

整体架构设计

先画一张图，让你对整体有个感觉。我个人习惯，做复杂系统前一定先画架构图，不然写着写着就乱了。

生产者(Producer) Kafka集群 消费者(Consumer) 消息处理流水线（消费者端） 责任链: 过滤 责任链: 转换 责任链: 校验 命令: 发送通知 命令: 持久化 命令: 回调 观察者: 通知所有订阅者 工厂: 创建处理器

从图里可以看到，消息从生产者进入Kafka，消费者拉取后，经过责任链处理，转成命令执行，最后通过观察者通知各个模块。工厂模式则负责把这些处理器、命令对象创建出来。嗯，这里要注意，工厂模式不是单独存在的，它贯穿在整个流水线中。

观察者模式：消息分发中枢

观察者模式说白了就是「发布-订阅」。Kafka本身就是一个分布式发布订阅系统，但我们在消费者端还需要二次分发——比如一条订单消息，既要通知库存系统，又要通知物流系统。

我在项目中遇到过一个问题：如果直接在消费逻辑里写死通知逻辑，每增加一个订阅方就要改代码，太痛苦了。用观察者模式就优雅多了。

// 观察者接口
public interface MessageObserver {
    void onMessage(MessageEvent event);
}

// 被观察者（主题）
public class MessageSubject {
    private List<MessageObserver> observers = new CopyOnWriteArrayList<>();
    
    public void attach(MessageObserver observer) {
        observers.add(observer);
    }
    
    public void detach(MessageObserver observer) {
        observers.remove(observer);
    }
    
    public void notifyObservers(MessageEvent event) {
        for (MessageObserver observer : observers) {
            observer.onMessage(event);
        }
    }
}

// 具体观察者：库存系统
public class InventoryObserver implements MessageObserver {
    @Override
    public void onMessage(MessageEvent event) {
        System.out.println("库存系统收到消息: " + event.getPayload());
        // 更新库存逻辑
    }
}

// 具体观察者：物流系统
public class LogisticsObserver implements MessageObserver {
    @Override
    public void onMessage(MessageEvent event) {
        System.out.println("物流系统收到消息: " + event.getPayload());
        // 安排配送逻辑
    }
}
避坑指南：我曾经在观察者列表里用了普通的ArrayList，结果并发环境下出现了ConcurrentModificationException。后来改成CopyOnWriteArrayList，问题解决。如果你对性能要求极高，可以考虑用Disruptor那样的无锁队列。

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

消息进来后，不能直接消费。得先过滤掉无效消息，再转换格式，最后校验完整性。这些步骤有先后顺序，而且未来可能增加新步骤。责任链模式正好适合这种场景。

你想想看，如果不用责任链，代码会写成什么样？一堆if-else嵌套，或者一个巨大的switch-case。维护起来简直噩梦。

// 责任链抽象类
public abstract class MessageHandler {
    protected MessageHandler nextHandler;
    
    public void setNext(MessageHandler handler) {
        this.nextHandler = handler;
    }
    
    public void handle(Message message) {
        if (canHandle(message)) {
            process(message);
        }
        if (nextHandler != null) {
            nextHandler.handle(message);
        }
    }
    
    protected abstract boolean canHandle(Message message);
    protected abstract void process(Message message);
}

// 过滤处理器
public class FilterHandler extends MessageHandler {
    @Override
    protected boolean canHandle(Message message) {
        return message.getType() != null && !message.getPayload().isEmpty();
    }
    
    @Override
    protected void process(Message message) {
        System.out.println("过滤通过: " + message.getId());
    }
}

// 转换处理器
public class TransformHandler extends MessageHandler {
    @Override
    protected boolean canHandle(Message message) {
        return true; // 所有消息都转换
    }
    
    @Override
    protected void process(Message message) {
        message.setPayload(message.getPayload().toUpperCase());
        System.out.println("转换完成: " + message.getPayload());
    }
}

// 校验处理器
public class ValidateHandler extends MessageHandler {
    @Override
    protected boolean canHandle(Message message) {
        return message.getPayload().length() > 0;
    }
    
    @Override
    protected void process(Message message) {
        System.out.println("校验通过: " + message.getId());
    }
}
核心要点：责任链的每个节点只关注自己的职责。过滤节点只做过滤，转换节点只做转换。这样新增一个「加密处理器」时，只需要新增一个类，不用改任何现有代码。

命令模式：封装消息动作

消息经过责任链处理后，需要执行具体动作。比如发送短信、写入数据库、调用外部API。这些动作如果直接写在消费逻辑里，耦合度太高。命令模式可以把每个动作封装成一个命令对象。

我记得有一次，业务方要求增加一个「延迟发送」功能。如果用命令模式，只需要新增一个DelayCommand，把原命令包装一下，加个定时器就行。改动非常小。

// 命令接口
public interface MessageCommand {
    void execute(Message message);
}

// 发送通知命令
public class NotifyCommand implements MessageCommand {
    private NotificationService notificationService;
    
    public NotifyCommand(NotificationService service) {
        this.notificationService = service;
    }
    
    @Override
    public void execute(Message message) {
        notificationService.send(message.getPayload());
    }
}

// 持久化命令
public class PersistCommand implements MessageCommand {
    private MessageRepository repository;
    
    public PersistCommand(MessageRepository repository) {
        this.repository = repository;
    }
    
    @Override
    public void execute(Message message) {
        repository.save(message);
    }
}

// 命令调用者（Invoker）
public class CommandInvoker {
    private List<MessageCommand> commands = new ArrayList<>();
    
    public void addCommand(MessageCommand command) {
        commands.add(command);
    }
    
    public void executeAll(Message message) {
        for (MessageCommand command : commands) {
            command.execute(message);
        }
    }
}

工厂模式：统一创建对象

前面提到的处理器、命令、观察者，如果都靠new来创建，代码里会散落一堆依赖。工厂模式可以把创建逻辑集中管理。而且，配合配置文件或注解，可以实现动态扩展。

我个人习惯用简单工厂+反射的方式，这样新增一个处理器时，只需要写类，然后在配置文件里注册一下就行。

// 处理器工厂
public class HandlerFactory {
    private static Map<String, MessageHandler> handlerMap = new HashMap<>();
    
    static {
        // 从配置文件加载处理器
        handlerMap.put("filter", new FilterHandler());
        handlerMap.put("transform", new TransformHandler());
        handlerMap.put("validate", new ValidateHandler());
    }
    
    public static MessageHandler getHandler(String type) {
        return handlerMap.get(type);
    }
    
    public static MessageHandler createChain(String... types) {
        MessageHandler head = null, current = null;
        for (String type : types) {
            MessageHandler handler = getHandler(type);
            if (head == null) {
                head = handler;
                current = handler;
            } else {
                current.setNext(handler);
                current = handler;
            }
        }
        return head;
    }
}

// 命令工厂
public class CommandFactory {
    public static MessageCommand createCommand(String type) {
        switch (type) {
            case "notify":
                return new NotifyCommand(new NotificationService());
            case "persist":
                return new PersistCommand(new MessageRepository());
            default:
                throw new IllegalArgumentException("Unknown command type: " + type);
        }
    }
}
注意：工厂模式虽然好用，但别滥用。如果一个对象的创建逻辑很简单（比如new一个POJO），就没必要用工厂。我见过有人给每个类都配一个工厂，结果代码量翻倍，维护成本反而更高。

整合：让四个模式协同工作

现在我们把四个模式串起来。消费者从Kafka拉取消息后，调用一个统一的入口方法。这个方法里，责任链处理消息，命令模式执行动作，观察者通知订阅方。工厂模式负责创建这些组件。

// 消息消费入口
public class MessageConsumer {
    private MessageHandler chain;
    private CommandInvoker invoker;
    private MessageSubject subject;
    
    public MessageConsumer() {
        // 使用工厂创建责任链
        this.chain = HandlerFactory.createChain("filter", "transform", "validate");
        
        // 使用工厂创建命令
        this.invoker = new CommandInvoker();
        invoker.addCommand(CommandFactory.createCommand("notify"));
        invoker.addCommand(CommandFactory.createCommand("persist"));
        
        // 注册观察者
        this.subject = new MessageSubject();
        subject.attach(new InventoryObserver());
        subject.attach(new LogisticsObserver());
    }
    
    public void consume(Message message) {
        // 1. 责任链处理
        chain.handle(message);
        
        // 2. 命令执行
        invoker.executeAll(message);
        
        // 3. 观察者通知
        subject.notifyObservers(new MessageEvent(message));
    }
}

你看，整个流程非常清晰。每个模式各司其职，互不干扰。如果将来要增加一个「消息去重」功能，只需要新增一个DeduplicateHandler，加到责任链里就行。如果增加一个「邮件通知」，新增一个EmailObserver注册到观察者列表里。

避坑总结

模式 常见坑 我的建议
观察者 观察者列表并发修改 用CopyOnWriteArrayList或Disruptor
责任链 链太长导致性能下降 控制链长度，超过5个考虑异步
命令 命令对象状态混乱 命令对象设计为无状态，或每次新建
工厂 工厂类变成上帝类 按领域拆分工厂，不要一个工厂管所有

最后说一句，设计模式不是银弹。我见过有人为了用模式而用模式，把简单问题复杂化。但在这个Kafka消息中间件的场景里，这四个模式确实能帮我们写出高内聚、低耦合的代码。如果你在实际项目中遇到类似的需求，不妨试试这个组合。

一句话总结：观察者管分发，责任链管处理，命令管执行，工厂管创建。四个模式各司其职，组合起来就是一套健壮的消息处理框架。

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

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