如何基于配置将数据存入DB或Kafka？对中介者模式存疑

一、更适配的策略模式实现 #

你的场景本质是根据配置选择不同的存储行为，策略模式比中介者模式更直接、易维护，先给这个实用方案：

步骤1：定义统一存储接口 #

先抽象出存储行为的标准，不管存DB还是Kafka都要实现这个接口：

public interface DataStorage {
    void save(Object data);
}

步骤2：实现具体存储逻辑 #

分别写DB和Kafka的实现类，各自负责自己的存储细节：

// DB存储实现
public class DbStorage implements DataStorage {
    @Override
    public void save(Object data) {
        // 这里写实际的DB操作，比如MyBatis/JDBC插入
        System.out.println("数据存入DB: " + data.toString());
    }
}

// Kafka存储实现
public class KafkaStorage implements DataStorage {
    @Override
    public void save(Object data) {
        // 这里写Kafka生产者发送逻辑，比如创建ProducerRecord发送
        System.out.println("数据存入Kafka: " + data.toString());
    }
}

步骤3：配置驱动的策略工厂 #

用工厂类根据配置值返回对应的存储实例：

import java.util.Map;

public class StorageFactory {
    // 预存所有存储策略
    private static final Map<String, DataStorage> STORAGE_MAP = Map.of(
        "1", new DbStorage(),
        "0", new KafkaStorage()
    );

    // 根据配置获取存储实例，默认返回Kafka
    public static DataStorage getStorage(String dbConfig) {
        return STORAGE_MAP.getOrDefault(dbConfig, new KafkaStorage());
    }
}

步骤4：业务代码调用 #

业务层只需要传入数据和配置，不需要关心底层存储逻辑：

public class BusinessHandler {
    public void handleData(Object data, String dbConfig) {
        // 先做业务处理，比如数据校验、转换
        DataStorage storage = StorageFactory.getStorage(dbConfig);
        storage.save(data);
    }
}

二、中介者模式实现方案 #

如果一定要用中介者模式（适合多组件交互复杂的场景），可以把配置、存储实例都交给中介者统一协调：

步骤1：定义中介者接口 #

中介者负责协调存储行为：

public interface StorageMediator {
    void saveData(Object data);
}

步骤2：实现配置驱动的中介者 #

中介者内部持有配置和两种存储实例，根据配置决定用哪个：

public class ConfigMediator implements StorageMediator {
    private final String dbConfig;
    private final DataStorage dbStorage;
    private final DataStorage kafkaStorage;

    public ConfigMediator(String dbConfig) {
        this.dbConfig = dbConfig;
        this.dbStorage = new DbStorage();
        this.kafkaStorage = new KafkaStorage();
    }

    @Override
    public void saveData(Object data) {
        if ("1".equals(dbConfig)) {
            dbStorage.save(data);
        } else {
            kafkaStorage.save(data);
        }
    }
}

步骤3：业务层调用 #

业务层通过中介者完成存储，不需要直接接触存储实现：

public class BusinessHandler {
    public void handleData(Object data, String dbConfig) {
        StorageMediator mediator = new ConfigMediator(dbConfig);
        mediator.saveData(data);
    }
}

三、额外建议 #

• 配置可以从配置中心（比如Nacos、Apollo）动态获取，不需要重启服务，只要在工厂/中介者里实时读取最新值就行
• 实际项目里可以用Spring的依赖注入代替硬编码new对象，比如把DbStorage和KafkaStorage注册成Bean，工厂直接注入
• 可以加异常降级逻辑，比如DB存储失败时自动切换到Kafka，保证数据不丢失

内容的提问来源于stack exchange，提问作者Rashhh