etcd配置中心实践 - 元素码农

发布时间: 2025-04-07 14:11

↑

# etcd配置中心实践

本文将详细介绍如何使用etcd实现配置中心功能，包括基本概念、实现方案和最佳实践。

## 配置中心概述

### 什么是配置中心

配置中心是分布式系统中的配置管理服务，它可以：

1. 集中管理各个系统的配置
2. 动态更新配置而无需重启服务
3. 保证配置的一致性
4. 提供配置修改的版本控制
5. 支持配置的环境隔离

### 为什么选择etcd

etcd作为配置中心的优势：

1. 强一致性保证配置的可靠性
2. 实时通知机制（Watch）
3. 版本控制支持
4. 简单的键值存储模型
5. 安全机制（RBAC、SSL/TLS）

## 实现方案

### 基本架构

### 配置存储

1. 配置结构设计
```go
type Config struct {
    Version     string      `json:"version"`
    Environment string      `json:"environment"`
    UpdateTime  time.Time   `json:"update_time"`
    Data        ConfigData  `json:"data"`
}

type ConfigData struct {
    Database struct {
        Host     string `json:"host"`
        Port     int    `json:"port"`
        Username string `json:"username"`
        Password string `json:"password"`
    } `json:"database"`
    Redis struct {
        Host     string `json:"host"`
        Port     int    `json:"port"`
        Password string `json:"password"`
    } `json:"redis"`
    App struct {
        LogLevel string `json:"log_level"`
        Port     int    `json:"port"`
    } `json:"app"`
}
```

2. 配置存储
```go
func saveConfig(client *clientv3.Client, env, app string, config *Config) error {
    key := fmt.Sprintf("/configs/%s/%s", env, app)
    data, err := json.Marshal(config)
    if err != nil {
        return fmt.Errorf("配置序列化失败: %v", err)
    }

// 使用事务确保原子性
    txn := client.Txn(context.Background())
    txn.If(
        clientv3.Compare(clientv3.Version(key), "=", 0),
    ).Then(
        clientv3.OpPut(key, string(data)),
    ).Else(
        clientv3.OpPut(key, string(data)),
    )

_, err = txn.Commit()
    if err != nil {
        return fmt.Errorf("保存配置失败: %v", err)
    }

return nil
}
```

### 配置读取

1. 获取配置
```go
func getConfig(client *clientv3.Client, env, app string) (*Config, error) {
    key := fmt.Sprintf("/configs/%s/%s", env, app)
    resp, err := client.Get(context.Background(), key)
    if err != nil {
        return nil, fmt.Errorf("获取配置失败: %v", err)
    }

if len(resp.Kvs) == 0 {
        return nil, fmt.Errorf("配置不存在")
    }

config := &Config{}
    err = json.Unmarshal(resp.Kvs[0].Value, config)
    if err != nil {
        return nil, fmt.Errorf("配置解析失败: %v", err)
    }

return config, nil
}
```

2. 监听配置变化
```go
func watchConfig(client *clientv3.Client, env, app string, callback func(*Config)) {
    key := fmt.Sprintf("/configs/%s/%s", env, app)
    rch := client.Watch(context.Background(), key)
    
    for wresp := range rch {
        for _, ev := range wresp.Events {
            if ev.Type == mvccpb.PUT {
                config := &Config{}
                err := json.Unmarshal(ev.Kv.Value, config)
                if err != nil {
                    log.Printf("配置解析失败: %v", err)
                    continue
                }
                callback(config)
            }
        }
    }
}
```

## 最佳实践

### 配置加密

1. 敏感信息加密
```go
func encryptValue(value string, key []byte) (string, error) {
    block, err := aes.NewCipher(key)
    if err != nil {
        return "", err
    }

ciphertext := make([]byte, aes.BlockSize+len(value))
    iv := ciphertext[:aes.BlockSize]
    if _, err := io.ReadFull(rand.Reader, iv); err != nil {
        return "", err
    }

stream := cipher.NewCFBEncrypter(block, iv)
    stream.XORKeyStream(ciphertext[aes.BlockSize:], []byte(value))

return base64.StdEncoding.EncodeToString(ciphertext), nil
}
```

2. 配置解密
```go
func decryptValue(encrypted string, key []byte) (string, error) {
    ciphertext, err := base64.StdEncoding.DecodeString(encrypted)
    if err != nil {
        return "", err
    }

block, err := aes.NewCipher(key)
    if err != nil {
        return "", err
    }

if len(ciphertext) < aes.BlockSize {
        return "", errors.New("密文太短")
    }

iv := ciphertext[:aes.BlockSize]
    ciphertext = ciphertext[aes.BlockSize:]

stream := cipher.NewCFBDecrypter(block, iv)
    stream.XORKeyStream(ciphertext, ciphertext)

return string(ciphertext), nil
}
```

### 配置热更新

1. 配置管理器
```go
type ConfigManager struct {
    client    *clientv3.Client
    config    atomic.Value
    callbacks []func(*Config)
    mu        sync.RWMutex
}

func (cm *ConfigManager) Start(env, app string) error {
    // 初始加载配置
    config, err := getConfig(cm.client, env, app)
    if err != nil {
        return err
    }
    cm.config.Store(config)

// 监听配置变化
    go watchConfig(cm.client, env, app, func(newConfig *Config) {
        cm.config.Store(newConfig)
        cm.notifyCallbacks(newConfig)
    })

return nil
}

func (cm *ConfigManager) GetConfig() *Config {
    return cm.config.Load().(*Config)
}

func (cm *ConfigManager) AddCallback(callback func(*Config)) {
    cm.mu.Lock()
    defer cm.mu.Unlock()
    cm.callbacks = append(cm.callbacks, callback)
}

func (cm *ConfigManager) notifyCallbacks(config *Config) {
    cm.mu.RLock()
    defer cm.mu.RUnlock()
    for _, callback := range cm.callbacks {
        go callback(config)
    }
}
```

### 配置版本控制

1. 版本管理
```go
type ConfigVersion struct {
    Version    string    `json:"version"`
    CreateTime time.Time `json:"create_time"`
    Data       Config    `json:"data"`
}

func saveConfigVersion(client *clientv3.Client, env, app string, config *Config) error {
    version := &ConfigVersion{
        Version:    uuid.New().String(),
        CreateTime: time.Now(),
        Data:       *config,
    }

key := fmt.Sprintf("/configs/%s/%s/versions/%s", env, app, version.Version)
    data, err := json.Marshal(version)
    if err != nil {
        return err
    }

_, err = client.Put(context.Background(), key, string(data))
    return err
}
```

2. 版本回滚
```go
func rollbackConfig(client *clientv3.Client, env, app, version string) error {
    // 获取指定版本的配置
    versionKey := fmt.Sprintf("/configs/%s/%s/versions/%s", env, app, version)
    resp, err := client.Get(context.Background(), versionKey)
    if err != nil {
        return err
    }

if len(resp.Kvs) == 0 {
        return fmt.Errorf("版本不存在")
    }

configVersion := &ConfigVersion{}
    err = json.Unmarshal(resp.Kvs[0].Value, configVersion)
    if err != nil {
        return err
    }

// 更新当前配置
    return saveConfig(client, env, app, &configVersion.Data)
}
```

## 应用示例

### 完整的配置中心客户端示例

```go
package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "time"
    "go.etcd.io/etcd/clientv3"
)

type ConfigClient struct {
    client  *clientv3.Client
    manager *ConfigManager
    env     string
    app     string
}

func NewConfigClient(endpoints []string, env, app string) (*ConfigClient, error) {
    client, err := clientv3.New(clientv3.Config{
        Endpoints:   endpoints,
        DialTimeout: 5 * time.Second,
    })
    if err != nil {
        return nil, err
    }

manager := &ConfigManager{
        client: client,
    }

return &ConfigClient{
        client:  client,
        manager: manager,
        env:     env,
        app:     app,
    }, nil
}

func (cc *ConfigClient) Start() error {
    return cc.manager.Start(cc.env, cc.app)
}

func (cc *ConfigClient) OnConfigChange(callback func(*Config)) {
    cc.manager.AddCallback(callback)
}

func (cc *ConfigClient) GetConfig() *Config {
    return cc.manager.GetConfig()
}

func main() {
    // 创建配置客户端
    client, err := NewConfigClient(
        []string{"localhost:2379"},
        "prod",
        "my-app",
    )
    if err != nil {
        log.Fatal(err)
    }

// 注册配置变更回调
    client.OnConfigChange(func(config *Config) {
        log.Printf("配置已更新: %+v\n", config)
    })

// 启动配置监听
    err = client.Start()
    if err != nil {
        log.Fatal(err)
    }

// 使用配置
    config := client.GetConfig()
    log.Printf("当前配置: %+v\n", config)

// 保持程序运行
    select {}
}
```

## 注意事项

1. 配置粒度
   - 避免过于频繁的配置更新
   - 合理划分配置项

2. 安全性
   - 敏感信息加密存储
   - 使用RBAC控制访问权限

3. 性能优化
   - 合理使用配置缓存
   - 避免过大的配置项

4. 容错处理
   - 配置加载失败时使用默认值
   - 实现配置回滚机制

## 总结

etcd作为配置中心具有以下优点：

1. 强一致性保证配置的可靠性
2. Watch机制支持配置热更新
3. 版本控制支持配置回滚
4. 安全机制保护敏感配置

在实际应用中，需要注意：

1. 合理设计配置结构
2. 实现可靠的配置热更新机制
3. 做好安全防护
4. 优化性能和资源使用