第4章：系统架构与设计决策

4.1 整体架构概览

4.1.1 系统层次架构

┌─────────────────────────────────────────────────────────────┐
│                     应用层 (API Layer)                        │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐     │
│  │ HTTP Server  │  │     Agent    │  │   Simulator   │     │
│  └──────────────┘  └──────────────┘  └──────────────┘     │
└─────────────────────────────────────────────────────────────┘
                              ↓
┌─────────────────────────────────────────────────────────────┐
│                   业务逻辑层 (Business Layer)                  │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐     │
│  │  MQTT    │ │  Auth    │ │  Alert   │ │  Sync    │     │
│  │  Client  │ │  Manager │ │ Notifier │ │ Manager  │     │
│  └──────────┘ └──────────┘ └──────────┘ └──────────┘     │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐     │
│  │ Monitor  │ │ Retention│ │Resource  │ │ Analyzer │     │
│  │          │ │  Manager │ │ Monitor  │ │          │     │
│  └──────────┘ └──────────┘ └──────────┘ └──────────┘     │
└─────────────────────────────────────────────────────────────┘
                              ↓
┌─────────────────────────────────────────────────────────────┐
│                    数据层 (Data Layer)                        │
│  ┌──────────────────────────────────────────────────────┐  │
│  │              Queue (本地磁盘队列)                      │  │
│  └──────────────────────────────────────────────────────┘  │
│  ┌──────────────────────────────────────────────────────┐  │
│  │              Database (sfsDb/LevelDB)                 │  │
│  └──────────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────────┘
                              ↓
┌─────────────────────────────────────────────────────────────┐
│                   外部系统 (External Systems)                  │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐     │
│  │ EdgeX Foundry│  │   MQTT Broker│  │   监控系统    │     │
│  └──────────────┘  └──────────────┘  └──────────────┘     │
└─────────────────────────────────────────────────────────────┘

4.1.2 核心模块依赖关系

// main.go 中的初始化顺序展示了模块依赖关系
func main() {
    // 1. 配置层（无依赖）
    appConfig, err = config.Load()
    
    // 2. 监控层（依赖配置）
    monitorInstance = monitor.NewMonitor()
    
    // 3. 告警层（依赖配置和监控）
    alertNotifier = alert.NewNotifier(appConfig)
    monitorInstance.SetNotifier(alertNotifier)
    
    // 4. 数据层（依赖配置）
    database.Init(appConfig.DBPath, ...)
    
    // 5. 队列层（无特殊依赖）
    dataQueue, err = queue.NewQueue("./data_queue")
    
    // 6. MQTT 层（依赖配置、队列、监控、分析器）
    mqttClient, err = mqtt.NewClient(appConfig, dataQueue, monitorInstance, analyzerInstance)
    
    // 7. HTTP 服务层（依赖数据层、配置、监控等）
    serverInstance := server.NewServer(database.Table, appConfig, ...)
}

4.2 设计决策详解

4.2.1 嵌入式数据库 vs 独立数据库

决策：使用嵌入式数据库（sfsDb/LevelDB）

理由：

1. 部署简单：无需额外部署数据库服务
2. 边缘友好：资源占用低，适合边缘设备
3. 性能优异：本地访问，无网络开销
4. 离线可用：不依赖网络连接

权衡：

维度	嵌入式数据库	独立数据库
部署复杂度	低	高
资源占用	低	高
并发能力	有限	强
分布式	不支持	支持
适用场景	边缘设备、单机应用	服务器、分布式系统

4.2.2 MQTT 作为数据传输协议

决策：使用 MQTT 协议接收 EdgeX 数据

理由：

1. 轻量级：协议开销小，适合带宽受限环境
2. 发布订阅：解耦数据生产者和消费者
3. EdgeX 原生支持：EdgeX Foundry 默认使用 MQTT
4. 可靠性：支持 QoS 0/1/2，保证消息送达

MQTT QoS 选择：

// mqtt/client.go 中的 QoS 设置
const DefaultQoS = 1 // 至少送达一次

// QoS 0：最多一次，可能丢失
// QoS 1：至少一次，保证送达，可能重复
// QoS 2：恰好一次，保证不重不丢，开销最大

4.2.3 本地磁盘队列作为缓冲层

决策：实现本地磁盘队列

理由：

1. 数据可靠性：数据库写入失败时不丢失数据
2. 削峰填谷：平滑突发流量
3. 离线缓冲：网络中断时缓存数据
4. 自动重试：后台自动处理队列数据

实现：

// queue/queue.go:32-47
type Queue struct {
    queueDir string
    mutex    sync.Mutex
}

func NewQueue(queueDir string) (*Queue, error) {
    if err := os.MkdirAll(queueDir, 0755); err != nil {
        return nil, fmt.Errorf("failed to create queue directory: %v", err)
    }
    return &Queue{
        queueDir: queueDir,
    }, nil
}

4.2.4 组合主键索引设计

决策：使用 (deviceName + timestamp) 作为组合主键

理由：

1. 查询模式匹配：常见查询是按设备和时间范围
2. 数据局部性：同一设备的数据物理存储在一起
3. 高效范围查询：利用组合主键前缀优化

设计示例：

// database/database.go:112-123
primaryKey, err := engine.DefaultPrimaryKeyNew("pk")
if err != nil {
    return fmt.Errorf("failed to create primary key: %v", err)
}
primaryKey.AddFields("deviceName", "timestamp")
if err := Table.CreateIndex(primaryKey); err != nil {
    // 忽略索引已存在的错误
    if err.Error() != "index already exists" {
        return fmt.Errorf("failed to create primary key index: %v", err)
    }
}

4.2.5 中间件模式设计

决策：使用中间件模式处理 HTTP 请求

理由：

1. 关注点分离：认证、日志、限流等独立处理
2. 可组合性：灵活组合不同中间件
3. 可维护性：每个中间件职责单一

实现示例：

// server/server.go:82-98
func DeviceNameMiddleware(next http.HandlerFunc) http.HandlerFunc {
    return func(w http.ResponseWriter, r *http.Request) {
        deviceName := r.URL.Query().Get("deviceName")
        if deviceName != "" {
            formattedDeviceName := common.FormatDeviceName(deviceName)
            url := *r.URL
            q := url.Query()
            q.Set("deviceName", formattedDeviceName)
            url.RawQuery = q.Encode()
            *r.URL = url
        }
        next(w, r)
    }
}

// 中间件组合使用
http.HandleFunc("/api/readings", auth.AuthMiddleware(DeviceNameMiddleware(s.handleQueryReadings)))

4.3 错误处理策略

4.3.1 分级错误处理

// 1. 致命错误：程序无法继续运行
if err = database.Init(appConfig.DBPath, ...); err != nil {
    log.Fatalf("Failed to initialize database: %v", err)
}

// 2. 严重错误：记录日志，继续运行
if err := alertNotifier.Start(); err != nil {
    log.Printf("Failed to start alert notifier: %v", err)
}

// 3. 可恢复错误：重试或降级
func BatchInsertWithRetry(...) error {
    for i := 0; i < maxRetries; i++ {
        _, err := tbl.BatchInsertNoInc(records)
        if err == nil {
            return nil
        }
        time.Sleep(retryInterval)
    }
    return fmt.Errorf("failed after %d attempts", maxRetries)
}

4.3.2 优雅关闭

// main.go:166-205
quit := make(chan os.Signal, 1)
signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
<-quit
log.Println("Shutting down adapter...")

// 按依赖逆序停止
if agentInstance != nil {
    agentInstance.Stop()
}
if retentionManager != nil {
    retentionManager.Stop()
}
if alertNotifier != nil {
    alertNotifier.Stop()
}

time.Sleep(5 * time.Second)
log.Println("Adapter exited")

4.4 配置管理设计

4.4.1 多来源配置支持

// 配置优先级（从低到高）
// 1. 默认值
// 2. 配置文件 (config.json)
// 3. 环境变量
// 4. 命令行参数

// config/config.go
type Config struct {
    DBPath             string `json:"db_path"`
    DBUseEncryption    bool   `json:"db_use_encryption"`
    DBEncryptionKey    string `json:"db_encryption_key"`
    MQTTBroker         string `json:"mqtt_broker"`
    HTTPPort           string `json:"http_port"`
    // ...
}

4.4.2 配置热更新

// server/server.go:1119-1185
func (s *Server) handleUpdateConfig(w http.ResponseWriter, r *http.Request) {
    var newConfig config.Config
    if err := json.NewDecoder(r.Body).Decode(&newConfig); err != nil {
        w.WriteHeader(http.StatusBadRequest)
        json.NewEncoder(w).Encode(map[string]string{"error": "Invalid request body"})
        return
    }
    configManager := config.GetConfigManager()
    if err := configManager.UpdateConfig(&newConfig); err != nil {
        w.WriteHeader(http.StatusInternalServerError)
        json.NewEncoder(w).Encode(map[string]string{"error": err.Error()})
        return
    }
}

4.5 实战练习

练习 4.1：架构设计评审

分析当前架构的优缺点，提出改进建议。

练习 4.2：设计决策模拟

假设有一个新的需求，分析并做出设计决策。

练习 4.3：错误处理改进

为某个模块设计更完善的错误处理策略。

4.6 本章小结

本章深入探讨了 sfsEdgeStore 的系统架构和设计决策：

• 整体架构的层次划分
• 关键设计决策的理由和权衡
• 错误处理策略的设计
• 配置管理的实现
• 优雅关闭机制

理解这些设计决策将帮助你更好地扩展和维护系统。

---

本书版本：1.0.0
最后更新：2026-03-08
sfsEdgeStore - 让边缘数据存储更简单！🚀
技术栈 - Go语言、sfsDb与EdgeX Foundry。纯golang工业物联网边缘计算技术栈
项目地址：GitHub
GitCode 镜像：GitCode