Golang模板方法模式与业务逻辑分离

package main

import (
    "fmt"
    "log"
    "time" // 增加一个时间模拟，让输出更真实
)

// DataProcessor 接口定义了数据处理的各个步骤
type DataProcessor interface {
    ReadData() ([]byte, error)
    ValidateData([]byte) error
    TransformData([]byte) ([]byte, error)
    SaveData([]byte) error
}

// ProcessDataTemplate 是模板方法，它定义了数据处理的固定流程
// 接受一个 DataProcessor 接口，具体实现由传入的处理器决定
func ProcessDataTemplate(p DataProcessor) error {
    fmt.Println("--- 开始数据处理流程 ---")

    // 步骤 1: 读取数据
    data, err := p.ReadData()
    if err != nil {
        return fmt.Errorf("读取数据失败: %w", err)
    }
    fmt.Printf("数据读取完成，大小: %d bytes\n", len(data))
    time.Sleep(100 * time.Millisecond) // 模拟耗时

    // 步骤 2: 校验数据
    err = p.ValidateData(data)
    if err != nil {
        return fmt.Errorf("数据校验失败: %w", err)
    }
    fmt.Println("数据校验通过")
    time.Sleep(100 * time.Millisecond) // 模拟耗时

    // 步骤 3: 转换数据
    transformedData, err := p.TransformData(data)
    if err != nil {
        return fmt.Errorf("数据转换失败: %w", err)
    }
    fmt.Printf("数据转换完成，新大小: %d bytes\n", len(transformedData))
    time.Sleep(100 * time.Millisecond) // 模拟耗时

    // 步骤 4: 保存数据
    err = p.SaveData(transformedData)
    if err != nil {
        return fmt.Errorf("数据保存失败: %w", err)
    }
    fmt.Println("数据保存成功")
    time.Sleep(100 * time.Millisecond) // 模拟耗时

    fmt.Println("--- 数据处理流程结束 ---")
    return nil
}

// --- 具体业务实现 ---

// CSVProcessor 处理CSV数据的具体实现
type CSVProcessor struct {
    source   string
    targetDB string
}

func NewCSVProcessor(source, targetDB string) *CSVProcessor {
    return &CSVProcessor{source: source, targetDB: targetDB}
}

func (c *CSVProcessor) ReadData() ([]byte, error) {
    // 模拟从CSV文件读取
    fmt.Printf("CSVProcessor: 从 '%s' 读取数据...\n", c.source)
    // 真实场景会读取文件内容
    return []byte("id,name,age\n1,Alice,30\n2,Bob,24"), nil
}

func (c *CSVProcessor) ValidateData(data []byte) error {
    // 模拟CSV数据校验，例如检查列数、数据类型等
    fmt.Println("CSVProcessor: 校验CSV数据格式和内容...")
    if len(data) < 20 { // 简单校验数据长度
        return fmt.Errorf("CSV数据过短，可能不完整")
    }
    // 实际中会解析CSV，检查每行每列
    return nil
}

func (c *CSVProcessor) TransformData(data []byte) ([]byte, error) {
    // 模拟CSV数据转换，例如解析为Go结构体，进行业务逻辑处理
    fmt.Println("CSVProcessor: 转换CSV数据为内部通用格式...")
    // 假设这里将CSV解析成JSON格式
    return []byte(`[{"id":1,"name":"Alice","age":30},{"id":2,"name":"Bob","age":24}]`), nil
}

func (c *CSVProcessor) SaveData(data []byte) error {
    // 模拟保存到数据库
    fmt.Printf("CSVProcessor: 保存转换后的数据到数据库 '%s'...\n", c.targetDB)
    // 实际中会执行数据库插入或更新操作
    return nil
}

// JSONProcessor 处理JSON数据的具体实现
type JSONProcessor struct {
    apiEndpoint string
    targetQueue string
}

func NewJSONProcessor(apiEndpoint, targetQueue string) *JSONProcessor {
    return &JSONProcessor{apiEndpoint: apiEndpoint, targetQueue: targetQueue}
}

func (j *JSONProcessor) ReadData() ([]byte, error) {
    // 模拟从API读取JSON
    fmt.Printf("JSONProcessor: 从 '%s' 读取JSON数据...\n", j.apiEndpoint)
    return []byte(`[{"user_id":101,"user_name":"Charlie"},{"user_id":102,"user_name":"David"}]`), nil
}

func (j *JSONProcessor) ValidateData(data []byte) error {
    // 模拟JSON数据校验，例如检查JSON Schema、字段完整性
    fmt.Println("JSONProcessor: 校验JSON数据结构和字段...")
    if len(data) < 30 { // 简单校验
        return fmt.Errorf("JSON数据过短，可能不完整")
    }
    // 实际中会进行JSON解析和结构验证
    return nil
}

func (j *JSONProcessor) TransformData(data []byte) ([]byte, error) {
    // 模拟JSON数据转换，例如调整字段名、合并数据
    fmt.Println("JSONProcessor: 转换JSON数据为内部通用格式...")
    // 假设这里将JSON解析后，转换为另一个JSON结构
    return []byte(`[{"id":101,"name":"Charlie"},{"id":102,"name":"David"}]`), nil
}

func (j *JSONProcessor) SaveData(data []byte) error {
    // 模拟发送到消息队列
    fmt.Printf("JSONProcessor: 发送转换后的数据到消息队列 '%s'...\n", j.targetQueue)
    // 实际中会向Kafka