第9讲：分布式配置中心

一、为什么需要配置中心

传统单体应用，配置文件写在本地，改配置需要重启服务。到了微服务时代，问题变得复杂：

问题场景：
┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐
│ 订单服务 │ │ 用户服务 │ │ 库存服务 │ │ 支付服务 │
│ 10 个实例 │ │ 8 个实例 │ │ 6 个实例 │ │ 4 个实例 │
├──────────┤ ├──────────┤ ├──────────┤ ├──────────┤
│ application.yml│ application.yml│ application.yml│ application.yml│
└──────────┘ └──────────┘ └──────────┘ └──────────┘

痛点：
1. 改一个数据库连接串，要登录 28 台机器修改
2. 改完还要逐个重启，容易漏掉
3. 配置错误导致线上事故，难以快速回滚
4. 敏感配置（密码、密钥）散落在各处

配置中心的核心能力：

• 集中管理：所有配置在一个地方维护
• 动态生效：修改配置无需重启服务
• 版本管理：配置变更可追溯、可回滚
• 权限控制：敏感配置加密存储

二、配置中心架构设计

┌─────────────────────────────────────────────────────────────────┐
│ 配置中心架构 │
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Admin UI │ │ CLI │ │ API │ ← 管理端 │
│ └─────┬────┘ └─────┬────┘ └─────┬────┘ │
│ │ │ │ │
│ ┌─────▼───────────────▼───────────────▼─────┐ │
│ │ Config Service │ │
│ │ ┌─────────────────────────────────────┐ │ │
│ │ │ 配置存储 (etcd/数据库) │ │ │
│ │ │ ┌──────┐ ┌──────┐ ┌──────┐ │ │ │
│ │ │ │应用A │ │应用B │ │应用C │ │ │ │
│ │ │ │v1..vn │ │v1..vn │ │v1..vn │ │ │ │
│ │ │ └──────┘ └──────┘ └──────┘ │ │ │
│ │ └─────────────────────────────────────┘ │ │
│ └───────────────────────────────────────────┘ │
│ │ │ │ │
│ ┌─────▼────┐ ┌─────▼────┐ ┌─────▼────┐ │
│ │ 服务A │ │ 服务B │ │ 服务C │ ← 客户端 │
│ │ 实例1-10 │ │ 实例1-8 │ │ 实例1-6 │ │
│ │ Watch │ │ Watch │ │ Watch │ │
│ └──────────┘ └──────────┘ └──────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘

三、配置模型定义

// distributed/config/model.go
package config

import (
"encoding/json"
"fmt"
"time"
)

// ConfigItem 配置项
type ConfigItem struct {
Key string `json:"key"`
Value string `json:"value"`
Group string `json:"group"` // 配置分组
Label string `json:"label"` // 标签（灰度用）
Format string `json:"format"` // 格式：text/json/yaml/properties
Encrypted bool `json:"encrypted"` // 是否加密
Description string `json:"description"`
Tags map[string]string `json:"tags,omitempty"`
}

// ConfigVersion 配置版本
type ConfigVersion struct {
Version int64 `json:"version"`
Items []ConfigItem `json:"items"`
ChangeLog string `json:"change_log"`
CreatedBy string `json:"created_by"`
CreatedAt time.Time `json:"created_at"`
}

// AppConfig 应用配置（客户端视角）
type AppConfig struct {
AppName string `json:"app_name"`
Env string `json:"env"` // dev/test/prod
Cluster string `json:"cluster"` // 集群标识
Version int64 `json:"version"`
Properties map[string]interface{} `json:"properties"`
UpdatedAt time.Time `json:"updated_at"`
}

// ConfigChangeEvent 配置变更事件
type ConfigChangeEvent struct {
AppName string
OldValue interface{}
NewValue interface{}
Key string
}

// ConfigWatcher 配置变更监听器
type ConfigWatcher interface {
OnChange(event ConfigChangeEvent)
}

四、配置存储实现（基于 etcd）

// distributed/config/store.go
package config

import (
"context"
"encoding/json"
"fmt"
"log"
"path"
"strings"
"sync"
"time"

clientv3 "go.etcd.io/etcd/client/v3"
)

// ConfigStore 配置存储
type ConfigStore struct {
client *clientv3.Client
basePath string // 基础路径，如 /config
watchers map[string][]ConfigWatcher
mu sync.RWMutex
}

// NewConfigStore 创建配置存储
func NewConfigStore(client *clientv3.Client, basePath string) *ConfigStore {
return &ConfigStore{
client: client,
basePath: strings.TrimRight(basePath, "/"),
watchers: make(map[string][]ConfigWatcher),
}
}

// SaveConfig 保存配置
func (s *ConfigStore) SaveConfig(ctx context.Context, appName, env string, items []ConfigItem) (*ConfigVersion, error) {
key := s.configKey(appName, env)

// 读取当前版本号
currentVersion, err := s.getCurrentVersion(ctx, key)
if err != nil {
return nil, err
}

newVersion := &ConfigVersion{
Version: currentVersion + 1,
Items: items,
ChangeLog: fmt.Sprintf("Updated at %s", time.Now().Format(time.RFC3339)),
CreatedBy: "admin",
CreatedAt: time.Now(),
}

data, err := json.Marshal(newVersion)
if err != nil {
return nil, fmt.Errorf("marshal config: %w", err)
}

// 保存新版本
versionKey := path.Join(key, fmt.Sprintf("v%d", newVersion.Version))
_, err = s.client.Put(ctx, versionKey, string(data))
if err != nil {
return nil, fmt.Errorf("save config: %w", err)
}

// 更新最新版本指针
latestKey := path.Join(key, "latest")
_, err = s.client.Put(ctx, latestKey, fmt.Sprintf("v%d", newVersion.Version))
if err != nil {
return nil, fmt.Errorf("update latest pointer: %w", err)
}

log.Printf("[Config] Saved config for %s/%s: version=%d", appName, env, newVersion.Version)
return newVersion, nil
}

// GetConfig 获取最新配置
func (s *ConfigStore) GetConfig(ctx context.Context, appName, env string) (*AppConfig, error) {
key := s.configKey(appName, env)

// 读取最新版本指针
latestKey := path.Join(key, "latest")
resp, err := s.client.Get(ctx, latestKey)
if err != nil {
return nil, fmt.Errorf("get latest pointer: %w", err)
}

if len(resp.Kvs) == 0 {
// 没有配置，返回默认
return &AppConfig{
AppName: appName,
Env: env,
Version: 0,
Properties: make(map[string]interface{}),
}, nil
}

versionName := string(resp.Kvs[0].Value)
versionKey := path.Join(key, versionName)

// 读取版本数据
resp, err = s.client.Get(ctx, versionKey)
if err != nil {
return nil, fmt.Errorf("get config version: %w", err)
}

if len(resp.Kvs) == 0 {
return nil, fmt.Errorf("config version not found: %s", versionKey)
}

var version ConfigVersion
if err := json.Unmarshal(resp.Kvs[0].Value, &version); err != nil {
return nil, fmt.Errorf("unmarshal config: %w", err)
}

// 构建 AppConfig
appConfig := &AppConfig{
AppName: appName,
Env: env,
Version: version.Version,
Properties: make(map[string]interface{}),
UpdatedAt: version.CreatedAt,
}

for _, item := range version.Items {
appConfig.Properties[item.Key] = item.Value
}

return appConfig, nil
}

// WatchConfig 监听配置变化
func (s *ConfigStore) WatchConfig(ctx context.Context, appName, env string) error {
key := s.configKey(appName, env)
watchKey := path.Join(key, "latest")

go func() {
watchChan := s.client.Watch(ctx, watchKey)
for watchResp := range watchChan {
for _, event := range watchResp.Events {
log.Printf("[Config] Config changed for %s/%s: %s", appName, env, event.Type)

// 重新加载配置
newConfig, err := s.GetConfig(ctx, appName, env)
if err != nil {
log.Printf("[Config] Reload failed: %v", err)
continue
}

// 通知监听器
s.notifyWatchers(appName, newConfig)
}
}
}()

return nil
}

// ListVersions 列出配置的所有版本
func (s *ConfigStore) ListVersions(ctx context.Context, appName, env string) ([]int64, error) {
key := s.configKey(appName, env)

resp, err := s.client.Get(ctx, key, clientv3.WithPrefix())
if err != nil {
return nil, err
}

versions := make([]int64, 0)
for _, kv := range resp.Kvs {
k := string(kv.Key)
if strings.HasPrefix(path.Base(k), "v") {
var ver int64
fmt.Sscanf(path.Base(k), "v%d", &ver)
versions = append(versions, ver)
}
}

return versions, nil
}

// Rollback 回滚到指定版本
func (s *ConfigStore) Rollback(ctx context.Context, appName, env string, targetVersion int64) error {
key := s.configKey(appName, env)

// 检查目标版本是否存在
versionKey := path.Join(key, fmt.Sprintf("v%d", targetVersion))
resp, err := s.client.Get(ctx, versionKey)
if err != nil {
return err
}
if len(resp.Kvs) == 0 {
return fmt.Errorf("version %d not found", targetVersion)
}

// 更新最新指针
latestKey := path.Join(key, "latest")
_, err = s.client.Put(ctx, latestKey, fmt.Sprintf("v%d", targetVersion))
if err != nil {
return fmt.Errorf("rollback: %w", err)
}

log.Printf("[Config] Rollback %s/%s to version %d", appName, env, targetVersion)
return nil
}

// RegisterWatcher 注册配置变更监听器
func (s *ConfigStore) RegisterWatcher(appName string, watcher ConfigWatcher) {
s.mu.Lock()
defer s.mu.Unlock()
s.watchers[appName] = append(s.watchers[appName], watcher)
}

// notifyWatchers 通知所有监听器
func (s *ConfigStore) notifyWatchers(appName string, config *AppConfig) {
s.mu.RLock()
watchers := s.watchers[appName]
s.mu.RUnlock()

for _, watcher := range watchers {
for key, value := range config.Properties {
watcher.OnChange(ConfigChangeEvent{
AppName: appName,
Key: key,
NewValue: value,
})
}
}
}

// configKey 生成 etcd 存储 key
func (s *ConfigStore) configKey(appName, env string) string {
return path.Join(s.basePath, appName, env)
}

// getCurrentVersion 获取当前版本号
func (s *ConfigStore) getCurrentVersion(ctx context.Context, key string) (int64, error) {
latestKey := path.Join(key, "latest")
resp, err := s.client.Get(ctx, latestKey)
if err != nil {
return 0, err
}

if len(resp.Kvs) == 0 {
return 0, nil
}

var version int64
fmt.Sscanf(string(resp.Kvs[0].Value), "v%d", &version)
return version, nil
}

五、客户端 SDK 实现

// distributed/config/client.go
package config

import (
"context"
"encoding/json"
"fmt"
"log"
"reflect"
"sync"
"time"

clientv3 "go.etcd.io/etcd/client/v3"
)

// ConfigClient 配置客户端
type ConfigClient struct {
store *ConfigStore
appConfig *AppConfig
listeners map[string][]ConfigChangeListener
mu sync.RWMutex
ctx context.Context
cancel context.CancelFunc
}

// ConfigChangeListener 配置变更回调
type ConfigChangeListener func(key string, oldValue, newValue interface{})

// NewConfigClient 创建配置客户端
func NewConfigClient(endpoints []string, basePath, appName, env string) (*ConfigClient, error) {
client, err := clientv3.New(clientv3.Config{
Endpoints: endpoints,
DialTimeout: 5 * time.Second,
})
if err != nil {
return nil, fmt.Errorf("connect etcd: %w", err)
}

store := NewConfigStore(client, basePath)
ctx, cancel := context.WithCancel(context.Background())

cc := &ConfigClient{
store: store,
listeners: make(map[string][]ConfigChangeListener),
ctx: ctx,
cancel: cancel,
}

// 初始加载配置
appConfig, err := store.GetConfig(ctx, appName, env)
if err != nil {
log.Printf("[Config] Initial load failed: %v", err)
appConfig = &AppConfig{
AppName: appName,
Env: env,
Properties: make(map[string]interface{}),
}
}
cc.appConfig = appConfig

// 启动监听
go cc.watchConfig(appName, env)

return cc, nil
}

// GetString 获取字符串配置
func (c *ConfigClient) GetString(key string, defaultValue string) string {
c.mu.RLock()
defer c.mu.RUnlock()

if val, ok := c.appConfig.Properties[key]; ok {
if str, ok := val.(string); ok {
return str
}
}
return defaultValue
}

// GetInt 获取整数配置
func (c *ConfigClient) GetInt(key string, defaultValue int) int {
c.mu.RLock()
defer c.mu.RUnlock()

if val, ok := c.appConfig.Properties[key]; ok {
switch v := val.(type) {
case float64:
return int(v)
case string:
var i int
fmt.Sscanf(v, "%d", &i)
return i
}
}
return defaultValue
}

// GetBool 获取布尔配置
func (c *ConfigClient) GetBool(key string, defaultValue bool) bool {
c.mu.RLock()
defer c.mu.RUnlock()

if val, ok := c.appConfig.Properties[key]; ok {
switch v := val.(type) {
case bool:
return v
case string:
return v == "true" || v == "1"
}
}
return defaultValue
}

// GetDuration 获取时间间隔配置
func (c *ConfigClient) GetDuration(key string, defaultValue time.Duration) time.Duration {
str := c.GetString(key, "")
if str == "" {
return defaultValue
}
d, err := time.ParseDuration(str)
if err != nil {
return defaultValue
}
return d
}

// GetObject 获取对象配置（JSON 格式）
func (c *ConfigClient) GetObject(key string, target interface{}) error {
c.mu.RLock()
defer c.mu.RUnlock()

val, ok := c.appConfig.Properties[key]
if !ok {
return fmt.Errorf("config key not found: %s", key)
}

data, err := json.Marshal(val)
if err != nil {
return err
}

return json.Unmarshal(data, target)
}

// OnChange 注册配置变更监听
func (c *ConfigClient) OnChange(key string, listener ConfigChangeListener) {
c.mu.Lock()
defer c.mu.Unlock()
c.listeners[key] = append(c.listeners[key], listener)
}

// watchConfig 监听配置变更
func (c *ConfigClient) watchConfig(appName, env string) {
key := c.store.configKey(appName, env)
watchKey := path.Join(key, "latest")

watchChan := c.store.client.Watch(c.ctx, watchKey)

for watchResp := range watchChan {
for range watchResp.Events {
// 重新加载配置
newConfig, err := c.store.GetConfig(c.ctx, appName, env)
if err != nil {
log.Printf("[Config] Reload failed: %v", err)
continue
}

c.applyChanges(newConfig)
}
}
}

// applyChanges 应用配置变更
func (c *ConfigClient) applyChanges(newConfig *AppConfig) {
c.mu.Lock()
defer c.mu.Unlock()

oldConfig := c.appConfig

// 检测变更并通知
for key, newVal := range newConfig.Properties {
oldVal := oldConfig.Properties[key]
if !reflect.DeepEqual(oldVal, newVal) {
// 通知监听器
if listeners, ok := c.listeners[key]; ok {
for _, listener := range listeners {
listener(key, oldVal, newVal)
}
}
}
}

c.appConfig = newConfig
log.Printf("[Config] Hot reload: version=%d", newConfig.Version)
}

// Close 关闭客户端
func (c *ConfigClient) Close() {
c.cancel()
c.store.client.Close()
}

六、配置热更新示例

// examples/config_demo.go
package main

import (
"fmt"
"log"
"time"

"distributed/config"
)

// DatabaseConfig 数据库配置
type DatabaseConfig struct {
Host string `json:"host"`
Port int `json:"port"`
Username string `json:"username"`
Password string `json:"password"`
MaxConns int `json:"max_conns"`
}

// RedisConfig Redis 配置
type RedisConfig struct {
Addrs []string `json:"addrs"`
Password string `json:"password"`
PoolSize int `json:"pool_size"`
}

func main() {
fmt.Println("=== 分布式配置中心演示 ===\\n")

// 模拟配置中心（使用内存存储）
fmt.Println("【场景1】配置管理\\n")

// 创建配置存储
client, _ := config.NewConfigClient(
[]string{"localhost:2379"},
"/config",
"order-service",
"prod",
)
defer client.Close()

// 初始配置
fmt.Println("加载初始配置…")
dbHost := client.GetString("db.host", "localhost")
dbPort := client.GetInt("db.port", 3306)
maxConns := client.GetInt("db.max_conns", 10)
debug := client.GetBool("debug.enabled", false)

fmt.Printf(" db.host = %s\\n", dbHost)
fmt.Printf(" db.port = %d\\n", dbPort)
fmt.Printf(" db.max_conns = %d\\n", maxConns)
fmt.Printf(" debug.enabled = %v\\n\\n", debug)

// 读取对象配置
fmt.Println("读取对象配置…")
var dbConfig DatabaseConfig
if err := client.GetObject("database", &dbConfig); err != nil {
fmt.Printf(" database config not found: %v\\n", err)
}

// 注册配置变更监听
fmt.Println("【场景2】配置热更新\\n")

client.OnChange("db.max_conns", func(key string, oldVal, newVal interface{}) {
fmt.Printf(" [热更新] 连接池大小变更: %v → %v\\n", oldVal, newVal)
})

client.OnChange("debug.enabled", func(key string, oldVal, newVal interface{}) {
fmt.Printf(" [热更新] 调试开关变更: %v → %v\\n", oldVal, newVal)
})

fmt.Println("监听配置变更中…（模拟外部修改配置）\\n")

// 模拟配置变更（实际生产中由 Admin 操作）
// 这里只是展示监听机制
fmt.Println("配置热更新演示完毕\\n")

// 配置版本管理
fmt.Println("【场景3】版本管理与回滚\\n")

fmt.Println("配置版本列表:")
versions, _ := client.ListVersions(context.Background(), "order-service", "prod")
for _, v := range versions {
fmt.Printf(" v%d\\n", v)
}

fmt.Println("\\n演示完成！")
}

七、配置加密

// distributed/config/crypto.go
package config

import (
"crypto/aes"
"crypto/cipher"
"crypto/rand"
"encoding/base64"
"fmt"
"io"
)

// ConfigEncryptor 配置加密器
type ConfigEncryptor struct {
key []byte
}

// NewConfigEncryptor 创建配置加密器
func NewConfigEncryptor(key string) *ConfigEncryptor {
// key 必须是 16/24/32 字节
k := []byte(key)
if len(k) < 16 {
padded := make([]byte, 16)
copy(padded, k)
k = padded
} else if len(k) < 24 {
padded := make([]byte, 24)
copy(padded, k)
k = padded
} else if len(k) < 32 {
padded := make([]byte, 32)
copy(padded, k)
k = padded
}

return &ConfigEncryptor{key: k[:32]}
}

// Encrypt 加密配置值
func (e *ConfigEncryptor) Encrypt(plaintext string) (string, error) {
block, err := aes.NewCipher(e.key)
if err != nil {
return "", err
}

aead, err := cipher.NewGCM(block)
if err != nil {
return "", err
}

nonce := make([]byte, aead.NonceSize())
if _, err := io.ReadFull(rand.Reader, nonce); err != nil {
return "", err
}

ciphertext := aead.Seal(nonce, nonce, []byte(plaintext), nil)
return base64.StdEncoding.EncodeToString(ciphertext), nil
}

// Decrypt 解密配置值
func (e *ConfigEncryptor) Decrypt(ciphertext string) (string, error) {
data, err := base64.StdEncoding.DecodeString(ciphertext)
if err != nil {
return "", err
}

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

aead, err := cipher.NewGCM(block)
if err != nil {
return "", err
}

nonceSize := aead.NonceSize()
if len(data) < nonceSize {
return "", fmt.Errorf("ciphertext too short")
}

nonce, ciphertext := data[:nonceSize], data[nonceSize:]
plaintext, err := aead.Open(nil, nonce, ciphertext, nil)
if err != nil {
return "", err
}

return string(plaintext), nil
}

八、生产坑位

#

坑位

现象

解决方案

九、本讲总结

┌──────────────────────────────────────────────────────────────────────────┐
│ 第9讲 核心收获 │
│ │
│ ✅ 理解了配置中心的核心价值： │
│ – 集中管理、动态生效、版本管理、权限控制 │
│ │
│ ✅ 实现了配置存储层： │
│ – etcd 存储配置项 │
│ – 版本管理（多版本存储） │
│ – 最新版本指针 │
│ │
│ ✅ 实现了客户端 SDK： │
│ – GetString/GetInt/GetBool/GetDuration/GetObject │
│ – OnChange 热更新回调 │
│ – Watch 机制实时感知变更 │
│ │
│ ✅ 实现了配置加密： │
│ – AES-GCM 加密 │
│ – 运行时透明解密 │
│ │
│ ✅ 覆盖了 6 个生产坑位与解决方案 │
│ │
└──────────────────────────────────────────────────────────────────────────┘

十、下一讲预告

第10讲（最终讲）：《分布式系统总结与实战》

你将学到：

• 分布式系统设计的十大原则
• 全系列知识点思维导图
• 一个完整的微服务 Demo（整合所有组件）
• 面试高频考点梳理
• 推荐书单与进阶路线

🧰 开发之余的小工具推荐

在调试配置中心的加密功能时，zz365.top 的 AES 加密工具可以快速验证加密/解密逻辑是否正确，确保敏感配置在传输和存储过程中的安全性。其 JSON 格式化工具则可以将复杂的配置结构展开为树形视图，便于检查配置项的层级关系。所有处理均在浏览器本地完成，不会泄露你的业务数据。