基于 Raft 与 Algolia 构建高可用分布式配置系统的 CQRS 架构实践

分布式系统

文章字数: 3.8k

阅读时长: 16 分

一个看似简单的分布式配置管理系统，在生产环境中，往往要同时满足两个相互冲突的核心需求：写入操作必须保证强一致性与高可用性，任何配置的变更都不能丢失或产生冲突；而读取与搜索操作则需要极致的速度、模糊匹配和高并发能力，以供成百上千的服务实例或开发者频繁查询。

传统的单一数据库方案，无论是关系型数据库还是通用 NoSQL，都很难在不引入大量复杂组件的前提下优雅地同时满足这两个极端。使用 PostgreSQL，可以通过 SERIALIZABLE 隔离级别保证写入一致性，但其全文搜索能力与 Algolia 这类专用引擎相比相去甚远。反之，若以 Elasticsearch 为核心，虽然搜索能力强大，但其作为强一致性数据存储的可靠性，尤其是在处理关键配置变更时，总让人心存顾虑。

这正是命令查询职责分离（CQRS）模式的用武之地。它允许我们为系统的写（Command）和读（Query）两侧选择最适合的技术栈，独立优化和扩展。

方案A：基于 Zookeeper/Etcd 的成熟方案

最直接的思路是使用 Etcd 这种成熟的、基于 Raft 的键值存储作为写模型。

优点:
- 开箱即用的强一致性保证。
- 社区成熟，稳定性经过大规模验证。
- 具备 Watch 机制，可以方便地监听数据变更。
缺点:
- 技术黑盒。当出现性能问题或需要深度定制时，我们对其内部机制的掌控力较弱。
- 资源占用相对较重，对于一个中等规模的配置中心而言可能有些过度设计。
- Watch 机制在需要同步到外部系统（如 Algolia）时，需要一个额外的、高可用的同步服务来消费事件流，这本身就引入了新的复杂性。

方案B：自研 Raft 日志核心与 Algolia 结合的 CQRS 架构

另一个方案是自己动手，围绕一个轻量级的 Raft 协议实现来构建写模型核心，并直接将已提交的日志条目同步到 Algolia。

优点:
- 对核心写入逻辑有完全的掌控力。可以精细控制状态机的行为、日志格式以及快照策略。
- 极致轻量。我们可以只实现所需的功能，避免引入庞大依赖。
- 数据同步逻辑与核心状态机耦合，可以保证任何被状态机成功应用（committed）的变更，都必然会进入同步队列，简化了架构。
缺点:
- 实现 Raft 协议有较高的技术门槛和风险，即使使用成熟的库（如 hashicorp/raft），也需要对协议有深刻理解。
- 需要自行处理日志存储、快照、成员变更等一系列工程问题。
- 存在最终一致性的延迟窗口，从写入成功到可在 Algolia 中被搜索到，会有毫秒到秒级的延迟。

对于一个追求技术掌控力和架构简洁性的团队来说，方案 B 更具吸引力。它将复杂性集中在可控的、自研的写模型核心中，而将读模型的复杂性完全外包给成熟的 SaaS 服务 Algolia。这是一种务实的权衡。

核心实现概览

我们将整个系统划分为四个主要部分：指令服务（Command Service）、查询服务（Query Service）、数据同步器（Data Synchronizer）以及前端应用（Frontend Application）。

graph TD
    subgraph "用户端"
        A[Nuxt.js + Lit 前端]
    end

    subgraph "服务端 API"
        B[API Gateway]
    end

    subgraph "写模型 (Command Side)"
        C[指令服务 Command Service]
        D[Raft Log]
        E[KV 状态机 FSM]
    end

    subgraph "读模型 (Query Side)"
        F[查询服务 Query Service]
        G[Algolia]
    end

    subgraph "同步链路"
        H[数据同步器 Data Synchronizer]
    end

    A -- "POST /api/configs (写)" --> B
    A -- "GET /api/search?q=... (读)" --> B

    B -- "写入请求" --> C
    B -- "查询请求" --> F

    C -- "Propose" --> D
    D -- "Replicate & Commit" --> D
    D -- "Apply" --> E

    E -- "Log Committed" --> H
    H -- "Index Data" --> G

    F -- "Search Query" --> G
    G -- "Search Results" --> F

Command 侧：基于 Raft 的日志与状态机

我们不从零造轮子，而是使用 hashicorp/raft 库来处理 Raft 协议的复杂性。我们的核心工作是实现其要求的 raft.FSM 接口，这便是我们的业务逻辑——一个简单的 KV 状态机。

以下是状态机和相关数据结构的核心 Go 代码。

fsm.go:

package main

import (
	"encoding/json"
	"fmt"
	"io"
	"sync"
	
	"github.com/hashicorp/raft"
	"github.com/tidwall/btree"
)

// CommandType 定义了可以对状态机执行的操作类型
type CommandType int

const (
	SetCommand CommandType = iota
	DeleteCommand
)

// Command 是将被应用到 Raft 日志中的基本操作单元
// 真实项目中，这里应该使用 Protobuf 等高效的序列化格式
type Command struct {
	Type  CommandType `json:"type"`
	Key   string      `json:"key"`
	Value string      `json:"value,omitempty"` // DeleteCommand 时为空
}

// KeyValueFSM 是我们实现的状态机，它处理 Raft 日志并更新内存中的 KV 存储
type KeyValueFSM struct {
	mu   sync.RWMutex
	data *btree.Map[string, string] // 使用 B-Tree 来保证有序，便于快照
	
	// 在状态机应用日志后，通过该通道通知同步器
	syncNotifier chan<- *Command 
}

// NewKeyValueFSM 创建一个新的状态机实例
// syncNotifier 是一个只写通道，用于解耦状态机和数据同步器
func NewKeyValueFSM(notifier chan<- *Command) *KeyValueFSM {
	return &KeyValueFSM{
		data:         btree.NewMap[string, string](),
		syncNotifier: notifier,
	}
}

// Apply 将一个 Raft 日志条目应用到状态机
// 这是 FSM 接口的核心方法，必须是确定性的
func (fsm *KeyValueFSM) Apply(log *raft.Log) interface{} {
	fsm.mu.Lock()
	defer fsm.mu.Unlock()

	var cmd Command
	if err := json.Unmarshal(log.Data, &cmd); err != nil {
		// 在生产环境中，这里应该有更健壮的错误处理和日志记录
		panic(fmt.Sprintf("failed to unmarshal command: %s", err.Error()))
	}

	switch cmd.Type {
	case SetCommand:
		fsm.data.Set(cmd.Key, cmd.Value)
	case DeleteCommand:
		fsm.data.Delete(cmd.Key)
	default:
		// 同样，需要健壮的错误处理
		panic(fmt.Sprintf("unrecognized command type: %d", cmd.Type))
	}

	// 关键步骤：在状态机成功应用变更后，将该变更发送给同步器
	// 使用非阻塞发送，避免同步器的缓慢处理阻塞 Raft 的状态机
	select {
	case fsm.syncNotifier <- &cmd:
	default:
		// 如果通道满了，意味着同步器严重滞后
		// 需要记录日志、告警，并考虑策略（如丢弃或阻塞）
		fmt.Printf("WARNING: Sync notifier channel is full. Command for key %s might be delayed.\n", cmd.Key)
	}

	return nil
}

// Snapshot 创建状态机的一个快照
// Raft 节点会定期调用此方法以压缩日志
func (fsm *KeyValueFSM) Snapshot() (raft.FSMSnapshot, error) {
	fsm.mu.RLock()
	defer fsm.mu.RUnlock()

	// 创建一个数据副本用于快照，避免并发问题
	clone := btree.NewMap[string, string]()
	fsm.data.Scan(func(key, value string) bool {
		clone.Set(key, value)
		return true
	})
	
	return &fsmSnapshot{data: clone}, nil
}

// Restore 从快照中恢复状态机状态
func (fsm *KeyValueFSM) Restore(rc io.ReadCloser) error {
	defer rc.Close()
	
	fsm.mu.Lock()
	defer fsm.mu.Unlock()

	// 简单起见，我们使用 JSON 进行快照序列化
	// 生产环境应考虑更高效的格式，如 Gob 或 Protobuf
	decoder := json.NewDecoder(rc)
	newData := btree.NewMap[string, string]()
	if err := decoder.Decode(&newData); err != nil {
		return fmt.Errorf("failed to decode snapshot: %w", err)
	}

	fsm.data = newData
	return nil
}

// fsmSnapshot 实现了 raft.FSMSnapshot 接口
type fsmSnapshot struct {
	data *btree.Map[string, string]
}

// Persist 将快照数据写入 sink
func (s *fsmSnapshot) Persist(sink raft.SnapshotSink) error {
	err := func() error {
		encoder := json.NewEncoder(sink)
		if err := encoder.Encode(s.data); err != nil {
			return fmt.Errorf("failed to encode snapshot data: %w", err)
		}
		return nil
	}()

	if err != nil {
		sink.Cancel()
	}
	
	// 在 return 之前必须调用 Close
	return sink.Close()
}

// Release 是一个空操作，因为我们的快照数据在内存中
func (s *fsmSnapshot) Release() {}

数据同步器：连接写模型与读模型

同步器的职责是监听 syncNotifier 通道，获取已提交的 Command，并调用 Algolia API 将变更应用到搜索索引中。这里的关键是错误处理和幂等性。

synchronizer.go:

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/algolia/algoliasearch-client-go/v3/algolia/opt"
	"github.com/algolia/algoliasearch-client-go/v3/algolia/search"
)

// AlgoliaObject 是我们存储在 Algolia 中的数据结构
type AlgoliaObject struct {
	ObjectID string `json:"objectID"` // Algolia 要求每个对象都有一个唯一的 objectID
	Key      string `json:"key"`
	Value    string `json:"value"`
}

// AlgoliaSynchronizer 负责将 FSM 的变更同步到 Algolia
type AlgoliaSynchronizer struct {
	index     *search.Index
	cmdChan   <-chan *Command
	ctx       context.Context
	batchSize int
	maxWait   time.Duration
}

// NewAlgoliaSynchronizer 创建同步器实例
func NewAlgoliaSynchronizer(ctx context.Context, client *search.Client, indexName string, cmdChan <-chan *Command) *AlgoliaSynchronizer {
	return &AlgoliaSynchronizer{
		index:     client.InitIndex(indexName),
		cmdChan:   cmdChan,
		ctx:       ctx,
		batchSize: 100, // 每100个变更批处理一次
		maxWait:   1 * time.Second, // 或者最多等待1秒
	}
}

// Run 启动同步循环
func (s *AlgoliaSynchronizer) Run() {
	ticker := time.NewTicker(s.maxWait)
	defer ticker.Stop()

	batch := make([]*Command, 0, s.batchSize)

	for {
		select {
		case <-s.ctx.Done():
			// 如果上下文取消，处理完最后一批数据后退出
			if len(batch) > 0 {
				s.processBatch(batch)
			}
			return
		case cmd, ok := <-s.cmdChan:
			if !ok { // 通道关闭
				if len(batch) > 0 {
					s.processBatch(batch)
				}
				return
			}
			batch = append(batch, cmd)
			if len(batch) >= s.batchSize {
				s.processBatch(batch)
				batch = make([]*Command, 0, s.batchSize) // 重置批次
			}
		case <-ticker.C:
			// 定时器触发，处理当前批次
			if len(batch) > 0 {
				s.processBatch(batch)
				batch = make([]*Command, 0, s.batchSize) // 重置批次
			}
		}
	}
}

// processBatch 处理一批命令，并推送到 Algolia
func (s *AlgoliaSynchronizer) processBatch(batch []*Command) {
	toAdd := make([]AlgoliaObject, 0)
	toDelete := make([]string, 0)

	for _, cmd := range batch {
		switch cmd.Type {
		case SetCommand:
			toAdd = append(toAdd, AlgoliaObject{
				ObjectID: cmd.Key, // 使用配置的 Key作为 Algolia 的 objectID，确保幂等性
				Key:      cmd.Key,
				Value:    cmd.Value,
			})
		case DeleteCommand:
			toDelete = append(toDelete, cmd.Key)
		}
	}
	
	// 在真实项目中，这里必须有带指数退避的重试逻辑
	if len(toAdd) > 0 {
		_, err := s.index.SaveObjects(toAdd, opt.AutoGenerateObjectIDIfNotExist(false))
		if err != nil {
			// 记录严重错误，并触发告警
			fmt.Printf("ERROR: Failed to save objects to Algolia: %v\n", err)
		}
	}
	if len(toDelete) > 0 {
		_, err := s.index.DeleteObjects(toDelete)
		if err != nil {
			fmt.Printf("ERROR: Failed to delete objects from Algolia: %v\n", err)
		}
	}

	fmt.Printf("Processed batch: %d adds, %d deletes\n", len(toAdd), len(toDelete))
}

一个常见的错误是：在 processBatch 中不处理部分失败。Algolia 的批量操作可能是部分成功的。生产级代码需要检查返回结果，对失败的部分进行重试。

API 设计：清晰的指令与查询边界

API 设计需要严格遵循 CQRS 原则。

POST /api/configs (Command):
- 职责: 接收写请求，将其提议给 Raft 集群，并等待日志被提交。
- Payload: { "key": "db.connection.string", "value": "..." }
- 响应: 成功时返回 202 Accepted 或 200 OK。202 更准确地表达了“请求已接受，但查询侧的更新是异步的”。返回 200 并附带提交的日志索引（log index）也是一种常见模式。
- 实现要点: 该接口的处理器必须是 Raft 集群的 Leader。如果请求发到 Follower 节点，需要返回重定向信息。hashicorp/raft 库提供了 Leader 的查询和订阅机制。
GET /api/search?q=<query> (Query):
- 职责: 接收读请求，将其代理给查询服务，该服务直接调用 Algolia。
- 响应: 200 OK，返回 Algolia 的搜索结果。
- 实现要点: 这个接口非常轻量，它不与 Raft 集群交互。它可以水平扩展，以应对高并发的查询负载。可以增加缓存层（如 Redis）来进一步降低对 Algolia 的请求压力。

Query 侧：Nuxt.js 与 Lit 的高性能前端

前端我们选择 Nuxt.js 作为应用框架，享受其 SSR 和文件路由带来的开发便利。对于性能敏感、可复用的 UI 部件，比如搜索框，我们使用 Lit。Lit 是一个构建轻量、快速的 Web Components 的库，它可以无缝集成到任何框架中。

components/config-search.js (Lit Component):

import { LitElement, html, css } from 'lit';

export class ConfigSearch extends LitElement {
  static styles = css`
    /* ... 一些基础样式 ... */
    :host { display: block; }
    .search-input { width: 100%; padding: 8px; font-size: 16px; }
    .results { margin-top: 10px; border: 1px solid #ccc; }
    .result-item { padding: 8px; border-bottom: 1px solid #eee; }
    .loading { padding: 8px; color: #888; }
  `;

  static properties = {
    results: { type: Array },
    isLoading: { type: Boolean },
    _debounceTimer: { state: true },
  };

  constructor() {
    super();
    this.results = [];
    this.isLoading = false;
    this._debounceTimer = null;
  }

  handleInput(e) {
    clearTimeout(this._debounceTimer);
    const query = e.target.value;

    if (!query) {
      this.results = [];
      return;
    }
    
    // 输入防抖，避免高频请求 API
    this.isLoading = true;
    this._debounceTimer = setTimeout(() => {
      this.fetchResults(query);
    }, 300); // 300ms 防抖延迟
  }

  async fetchResults(query) {
    try {
      // 在 Nuxt 中, API 路由会自动代理
      const response = await fetch(`/api/search?q=${encodeURIComponent(query)}`);
      if (!response.ok) {
        throw new Error(`HTTP error! status: ${response.status}`);
      }
      const data = await response.json();
      this.results = data.hits || [];
    } catch (error) {
      console.error("Failed to fetch search results:", error);
      this.results = []; // 出错时清空结果
    } finally {
      this.isLoading = false;
    }
  }

  render() {
    return html`
      <input
        type="search"
        class="search-input"
        placeholder="Search configurations..."
        @input=${this.handleInput}
      />
      <div class="results">
        ${this.isLoading ? html`<div class="loading">Loading...</div>` : ''}
        ${!this.isLoading && this.results.map(item => html`
          <div class="result-item">
            <strong>${item.key}</strong>: <pre>${item.value}</pre>
          </div>
        `)}
        ${!this.isLoading && this.results.length === 0 ? html`<div class="loading">No results.</div>` : ''}
      </div>
    `;
  }
}

customElements.define('config-search', ConfigSearch);

在 Nuxt 页面中使用它 (pages/index.vue)，需要确保在客户端渲染时才引入该组件，因为 Web Components 依赖浏览器 API。

<template>
  <div>
    <h1>Configuration Management</h1>
    <ClientOnly>
      <config-search></config-search>
    </ClientOnly>
  </div>
</template>

<script setup>
import { onMounted } from 'vue';

// 在 onMounted 钩子中动态导入 Lit 组件，确保在客户端执行
onMounted(async () => {
  if (process.client) {
    await import('~/components/config-search.js');
  }
});
</script>

这里的核心在于，前端组件完全不关心数据是如何写入或保持一致的。它只与一个高性能、最终一致的查询端点交互，这使得前端开发和用户体验都变得极为简单和流畅。

架构的扩展性与局限性

此架构的扩展性体现在读写两侧的完全解耦。如果写入压力增大，我们可以为 Raft 集群增加节点或优化磁盘 I/O。如果查询量暴增，我们只需水平扩展无状态的 Query Service 实例，Algolia 本身就是为高并发查询而设计的。

然而，这个方案并非没有局限性。首先，我们自担了维护 Raft 集群的运维成本，包括节点替换、成员变更、快照管理和灾难恢复。虽然 hashicorp/raft 简化了协议实现，但工程实践依然充满挑战。其次，数据从写入到可被搜索到的延迟是固有存在的，这个延迟窗口（propagation delay）取决于同步器的处理能力和网络状况，对于需要“读己之写”（read-your-writes）一致性的场景可能不适用。最后，对 Algolia 的依赖构成了厂商锁定，尽管其服务质量很高，但这在架构决策中仍是一个需要考虑的因素。未来的迭代方向可能包括将数据同步器本身设计为高可用的分布式服务，以及引入更精细化的监控，来度量和告警整个数据链路的端到端延迟。

Raft 协议 API 设计 Algolia Nuxt.js Lit

构建代码审查流程的可观测性管道：从 CircleCI Webhook 到 Grafana 面板

2023-10-27 DevOps

CircleCI Grafana Code Review 可观测性

使用 Actix-web 与 ZeroMQ 构建一个进程隔离的分布式插件系统并通过 GitHub Actions 实现自动化部署

2023-10-27 后端架构

Actix-web ZeroMQ GitHub Actions Rust 微内核架构