构建基于Consul领导者选举的高可用时序数据聚合器及其单元测试实践

分布式系统

文章字数: 4.9k

阅读时长: 21 分

在一个典型的时序数据收集中间件集群中，多个节点同时接收来自不同数据源的指标。直接将这些原始数据写入后端的时序数据库（TSDB）会带来巨大的写入压力和存储成本。一个常见的优化是引入一个聚合层：集群中仅有一个节点作为“领导者”（Leader），负责对一个时间窗口内的数据进行预聚合（如计算均值、P99分位数等），然后将聚合后的结果写入TSDB。其他节点则作为“追随者”（Follower），仅转发数据或处于待命状态。

这个架构的核心挑战在于如何保证聚合节点的高可用性。当领导者节点因故障宕机或网络隔离时，集群必须能迅速、无歧义地选举出新的领导者接替其工作。任何延迟或选举失败都意味着数据聚合的中断。更糟糕的是，如果出现“脑裂”（Split-Brain），即两个节点都认为自己是领导者，就会导致数据被重复聚合，污染下游的TSDB。

我们需要的不是一个重量级的共识协议实现，而是一个轻量且可靠的分布式锁服务来仲裁领导权。Consul，作为服务发现和配置工具，其内置的Session和KV存储提供了一个近乎完美的领导者选举原语。

本文的目标是，使用Go语言和Consul，从零开始构建一个健壮的时序数据聚合器。重点不在于聚合算法本身，而在于如何实现其背后可靠的、自愈的领导者选举循环，以及如何为这个高度依赖外部网络和状态的分布式组件编写确定性的、可隔离的单元测试。

核心设计：状态机与调谐循环

聚合器节点本质上是一个状态机，它可以在几个状态之间转换：IDLE（空闲）、ACQUIRING（尝试获取领导权）、LEADER（持有领导权并执行聚合任务）、FOLLOWER（未持有领导权，待命中）。

stateDiagram-v2
    [*] --> IDLE: 启动

    IDLE --> ACQUIRING: 开始选举循环
    ACQUIRING --> LEADER: 成功获取锁
    ACQUIRING --> FOLLOWER: 获取锁失败 (锁已被占用)
    ACQUIRING --> IDLE: 发生不可恢复错误 (如Consul连接断开)

    LEADER --> LEADER: 周期性刷新Session/锁
    LEADER --> FOLLOWER: 主动释放锁或Session失效
    LEADER --> IDLE: 发生不可恢复错误

    FOLLOWER --> ACQUIRING: 监测到锁被释放，重新尝试获取
    FOLLOWER --> FOLLOWER: 周期性监测锁状态
    FOLLOWER --> IDLE: 发生不可恢复错误

整个过程由一个无限的“调谐循环”（Reconciliation Loop）驱动，它不断尝试将节点的当前状态调整为期望状态（即成为领导者）。

让我们先定义聚合器的核心结构。它需要一个Consul客户端、一个唯一的节点ID、以及一些用于控制状态的内部字段。

// aggregator.go

package aggregator

import (
	"context"
	"fmt"
	"log"
	"sync"
	"time"

	"github.com/hashicorp/consul/api"
)

const (
	DefaultSessionTTL      = 15 * time.Second
	DefaultLockAcquireWait = 5 * time.Second
	DefaultLockMonitorRate = 2 * time.Second
)

// Aggregator 负责执行领导者选举和在成为领导者时运行聚合任务
type Aggregator struct {
	nodeID      string
	lockKey     string
	consulClient *api.Client

	// 内部状态
	mu         sync.Mutex
	isLeader   bool
	sessionID  string
	stopCh     chan struct{} // 用于优雅停止
	wg         sync.WaitGroup

	// 用于注入模拟任务，方便测试
	aggregationTask func(ctx context.Context)
}

// NewAggregator 创建一个新的聚合器实例
func NewAggregator(nodeID, lockKey, consulAddr string) (*Aggregator, error) {
	conf := api.DefaultConfig()
	conf.Address = consulAddr
	client, err := api.NewClient(conf)
	if err != nil {
		return nil, fmt.Errorf("failed to create consul client: %w", err)
	}

	return &Aggregator{
		nodeID:      nodeID,
		lockKey:     lockKey,
		consulClient: client,
		stopCh:      make(chan struct{}),
		// 默认任务是一个空操作
		aggregationTask: func(ctx context.Context) {
			log.Printf("[%s] performing default aggregation task", nodeID)
			select {
			case <-ctx.Done():
				return
			case <-time.After(1 * time.Second): // 模拟工作负载
			}
		},
	}, nil
}

// Start 启动领导者选举循环
func (a *Aggregator) Start() {
	a.wg.Add(1)
	go a.leaderElectionLoop()
}

// Stop 优雅地停止聚合器
func (a *Aggregator) Stop() {
	close(a.stopCh)
	a.wg.Wait()
	log.Printf("[%s] Aggregator stopped.", a.nodeID)
}

// IsLeader 返回当前节点是否为领导者
func (a *Aggregator) IsLeader() bool {
	a.mu.Lock()
	defer a.mu.Unlock()
	return a.isLeader
}

func (a *Aggregator) setLeader(status bool) {
	a.mu.Lock()
	defer a.mu.Unlock()
	if a.isLeader != status {
		a.isLeader = status
		log.Printf("[%s] Leadership status changed to: %v", a.nodeID, status)
	}
}

领导者选举的实现细节

Consul的领导者选举依赖于两个关键组件：Session 和 KV Store Lock。

Session: 客户端创建一个Session，它有一个关联的TTL（Time-To-Live）。客户端必须在TTL到期前周期性地更新Session来“续命”，证明自己仍然存活。如果客户端宕机或网络中断，无法更新Session，Consul会自动销毁该Session。
KV Store Lock: KV存储的Acquire操作可以关联一个Session。一个Key只能被一个Session锁定。当持有锁的Session被销毁时，这个锁会自动释放，允许其他客户端来获取。

这正是我们需要的机制。我们的调谐循环将执行以下步骤：

创建一个Session。
在一个无限循环中，尝试用该Session去获取指定的lockKey。
如果成功获取锁，则将自身状态设为LEADER，并启动聚合任务。同时，持续监控锁的状态，因为我们可能因为Session失效而“非自愿地”失去锁。
如果获取失败，则进入FOLLOWER状态，等待一段时间后再次尝试。
在整个过程中，后台必须有一个goroutine负责更新Session，保持其活性。

// aggregator.go (续)

// leaderElectionLoop 是核心的调谐循环
func (a *Aggregator) leaderElectionLoop() {
	defer a.wg.Done()
	log.Printf("[%s] Starting leader election loop for key '%s'", a.nodeID, a.lockKey)

	// 循环直到被外部停止
	for {
		select {
		case <-a.stopCh:
			// 如果当前是领导者，尝试优雅释放锁
			if a.IsLeader() && a.sessionID != "" {
				a.releaseLockAndDestroySession()
			}
			return
		default:
			// 如果没有session，创建一个
			if a.sessionID == "" {
				sessionID, err := a.createSession()
				if err != nil {
					log.Printf("[%s] Error creating session: %v. Retrying in %s...", a.nodeID, err, DefaultLockAcquireWait)
					time.Sleep(DefaultLockAcquireWait)
					continue
				}
				a.sessionID = sessionID
				// 启动session续期goroutine
				go a.renewSession(a.sessionID)
			}
			
			// 尝试获取锁
			acquired, err := a.acquireLock()
			if err != nil {
				log.Printf("[%s] Error acquiring lock: %v. Retrying...", a.nodeID, err)
				// 发生错误时，销毁当前session，下一轮循环会重建
				a.destroySession(a.sessionID)
				a.sessionID = ""
				time.Sleep(DefaultLockAcquireWait)
				continue
			}

			if acquired {
				// 成功获取锁，成为领导者
				a.runAsLeader()
			} else {
				// 未获取到锁，成为追随者
				a.runAsFollower()
			}
		}
	}
}

// createSession 在Consul中创建一个新的session
func (a *Aggregator) createSession() (string, error) {
	sessionEntry := &api.SessionEntry{
		Name:      fmt.Sprintf("aggregator-leader-%s", a.nodeID),
		TTL:       DefaultSessionTTL.String(),
		Behavior:  api.SessionBehaviorDelete, // Session失效时，关联的锁自动释放
	}
	sessionID, _, err := a.consulClient.Session().Create(sessionEntry, nil)
	if err != nil {
		return "", err
	}
	log.Printf("[%s] Created Consul session: %s", a.nodeID, sessionID)
	return sessionID, nil
}

// renewSession 周期性地为session续期
func (a *Aggregator) renewSession(sessionID string) {
	ctx, cancel := context.WithCancel(context.Background())
	go func() {
		<-a.stopCh
		cancel()
	}()
	
	// 在一个独立的goroutine中执行续期，以防主循环阻塞
	// Done() channel会在session失效或被销毁时关闭
	err := a.consulClient.Session().RenewPeriodic(DefaultSessionTTL.String(), sessionID, nil, ctx.Done())
	if err != nil {
		log.Printf("[%s] Session renewal failed for session %s: %v", a.nodeID, sessionID, err)
	}
	log.Printf("[%s] Session renewal stopped for session %s", a.nodeID, sessionID)
}

// acquireLock 尝试获取分布式锁
func (a *Aggregator) acquireLock() (bool, error) {
	kvPair := &api.KVPair{
		Key:     a.lockKey,
		Value:   []byte(a.nodeID),
		Session: a.sessionID,
	}
	acquired, _, err := a.consulClient.KV().Acquire(kvPair, nil)
	return acquired, err
}

// runAsLeader 作为领导者运行，执行聚合任务并监控锁状态
func (a *Aggregator) runAsLeader() {
	a.setLeader(true)
	
	taskCtx, taskCancel := context.WithCancel(context.Background())
	defer taskCancel()
	
	go a.aggregationTask(taskCtx)

	// 监控锁状态，如果锁丢失（例如session失效），则退出leader角色
	// 这里使用阻塞查询，当key的状态改变时，查询会立即返回
	for {
		select {
		case <-a.stopCh:
			a.releaseLockAndDestroySession()
			return
		default:
			kv, _, err := a.consulClient.KV().Get(a.lockKey, &api.QueryOptions{WaitTime: DefaultLockMonitorRate})
			if err != nil {
				log.Printf("[%s] Error monitoring lock key: %v. Stepping down.", a.nodeID, err)
				a.setLeader(false)
				return
			}
			// 如果key不存在，或持有锁的session不是我们自己的，说明我们已失去领导权
			if kv == nil || kv.Session != a.sessionID {
				log.Printf("[%s] Lost leadership. Current lock holder: %s", a.nodeID, kv.Session)
				a.setLeader(false)
				// Session可能已经失效，销毁它以防万一，并在主循环中重建
				a.destroySession(a.sessionID)
				a.sessionID = ""
				return
			}
			// 锁仍然由我们持有，继续循环监控
		}
	}
}

// runAsFollower 作为追随者运行，等待锁被释放
func (a *Aggregator) runAsFollower() {
	a.setLeader(false)
	
	var lastIndex uint64 = 0
	for {
		select {
		case <-a.stopCh:
			return
		default:
			// 使用阻塞查询等待锁key的变化
			// 如果锁被释放(kv.Session == "")，这个查询会返回
			kv, meta, err := a.consulClient.KV().Get(a.lockKey, &api.QueryOptions{WaitIndex: lastIndex})
			if err != nil {
				log.Printf("[%s] Error waiting for lock release: %v. Retrying...", a.nodeID, err)
				time.Sleep(DefaultLockMonitorRate)
				continue
			}
			
			// 如果WaitIndex没有变化，说明是超时返回，继续等待
			if meta.LastIndex == lastIndex {
				continue
			}
			lastIndex = meta.LastIndex
			
			// 如果key不存在或没有session持有它，说明锁已释放，我们可以退出follower状态去尝试获取锁
			if kv == nil || kv.Session == "" {
				log.Printf("[%s] Lock appears to be free. Re-entering election.", a.nodeID)
				return // 返回到主循环以尝试获取锁
			}
			// 否则，锁仍然被其他节点持有，继续等待
		}
	}
}

// releaseLockAndDestroySession 主动释放锁并销毁session
func (a *Aggregator) releaseLockAndDestroySession() {
	log.Printf("[%s] Releasing lock and destroying session %s", a.nodeID, a.sessionID)
	kvPair := &api.KVPair{
		Key:     a.lockKey,
		Session: a.sessionID,
	}
	// 释放锁
	a.consulClient.KV().Release(kvPair, nil)
	// 销毁session
	a.destroySession(a.sessionID)
	a.sessionID = ""
	a.setLeader(false)
}

// destroySession 销毁一个指定的session
func (a *Aggregator) destroySession(sessionID string) {
	if sessionID != "" {
		a.consulClient.Session().Destroy(sessionID, nil)
	}
}

这段代码已经构成了一个功能完备的领导者选举组件。但在真实项目中，这样的代码是难以交付的，因为它有一个致命缺陷：极差的可测试性。Aggregator结构体与github.com/hashicorp/consul/api包紧密耦合。要测试它的逻辑，我们必须在测试环境中启动一个真实的Consul实例，这会带来诸多问题：

测试环境复杂：需要管理Consul进程的生命周期。
测试速度慢：网络通信和外部进程调用会显著拖慢测试执行速度。
测试不稳定：测试结果会受到网络抖动、Consul实例状态等非确定性因素的影响。
难以模拟边缘情况：如何精确地模拟“Session突然失效”、“KV写入超时”或“网络分区”等故障场景？

重构以实现可测试性

解决之道在于依赖倒置。我们不应该直接依赖Consul的具体实现，而应该依赖一个我们自己定义的、描述了所需行为的接口。

首先，我们识别出Aggregator与Consul交互的关键操作：创建/销毁Session、更新Session、获取/释放KV锁、监控KV变化。我们将这些操作抽象成接口。

// consul_adapter.go

package aggregator

import (
	"context"
	"github.com/hashicorp/consul/api"
)

// ConsulLock 接口定义了与Consul进行锁操作的最小行为集合
type ConsulLock interface {
	CreateSession(se *api.SessionEntry) (string, error)
	DestroySession(sessionID string) error
	RenewSessionPeriodic(sessionID string, doneCh <-chan struct{}) error
	AcquireLock(p *api.KVPair) (bool, error)
	ReleaseLock(p *api.KVPair) (bool, error)
	MonitorLock(key string, stopCh <-chan struct{}) <-chan *api.KVPair
}

接着，我们为真实的Consul API客户端创建一个适配器，实现这个接口。

// consul_adapter.go (续)

type consulAPILock struct {
	client *api.Client
}

func newConsulAPILock(client *api.Client) ConsulLock {
	return &consulAPILock{client: client}
}

func (c *consulAPILock) CreateSession(se *api.SessionEntry) (string, error) {
	id, _, err := c.client.Session().Create(se, nil)
	return id, err
}

func (c *consulAPILock) DestroySession(sessionID string) error {
	_, err := c.client.Session().Destroy(sessionID, nil)
	return err
}

func (c *consulAPILock) RenewSessionPeriodic(sessionID string, doneCh <-chan struct{}) error {
	return c.client.Session().RenewPeriodic(DefaultSessionTTL.String(), sessionID, nil, doneCh)
}

func (c *consulAPILock) AcquireLock(p *api.KVPair) (bool, error) {
	acquired, _, err := c.client.KV().Acquire(p, nil)
	return acquired, err
}

func (c *consulAPILock) ReleaseLock(p *api.KVPair) (bool, error) {
	released, _, err := c.client.KV().Release(p, nil)
	return released, err
}

func (c *consulAPILock) MonitorLock(key string, stopCh <-chan struct{}) <-chan *api.KVPair {
	// 真实监控逻辑比较复杂，这里为了演示简化。
	// 生产级代码需要处理阻塞查询的完整逻辑。
	// 但对于测试来说，这个接口的模拟实现才是关键。
	// 在重构后的Aggregator中，我们不会直接调用这个，而是将监控逻辑保留在内部。
	// 这里的接口设计主要是为了演示依赖倒置。
	// 一个更完整的重构会将所有外部交互都通过接口进行。
	// 为保持文章重点，我们将专注于测试核心选举逻辑，这需要模拟Acquire/Release/Session。
	panic("MonitorLock in real adapter is complex; focus is on mock for tests")
}

现在，重构Aggregator，使其接收我们定义的ConsulLock接口，而不是具体的*api.Client。这对外部使用者来说几乎是透明的，但在内部，所有的Consul调用都将通过这个接口。

为了简洁，以下只展示修改后的Aggregator构造函数和字段，其余逻辑中将a.consulClient.XXX替换为a.consulLock.XXX即可。

// aggregator_refactored.go (部分)

// ...
type Aggregator struct {
    // ...
    // consulClient *api.Client // -> 被替换
    consulLock ConsulLock
    // ...
}

// NewAggregator 创建一个新的聚合器实例
func NewAggregatorWithLock(nodeID, lockKey string, lockImpl ConsulLock) (*Aggregator, error) {
	// ...
    return &Aggregator{
		nodeID:      nodeID,
		lockKey:     lockKey,
		consulLock:  lockImpl, // 注入接口实现
		// ...
	}, nil
}

编写确定性的单元测试

重构完成后，最激动人心的部分来了：为这个复杂的分布式组件编写单元测试。我们将创建一个MockConsulLock，它完全在内存中运行，并允许我们在测试代码中精确地控制其行为。

// aggregator_test.go

package aggregator

import (
	"context"
	"errors"
	"sync"
	"testing"
	"time"

	"github.com/hashicorp/consul/api"
	"github.com/stretchr/testify/assert"
)

// mockConsulLock 是 ConsulLock 接口的模拟实现，用于单元测试
type mockConsulLock struct {
	mu          sync.Mutex
	lockHolder  string
	session     string
	failAcquire bool
	failSession bool
	lockEvents  chan *api.KVPair
}

func newMockConsulLock() *mockConsulLock {
	return &mockConsulLock{
		lockEvents: make(chan *api.KVPair, 1),
	}
}

// 控制模拟行为的方法
func (m *mockConsulLock) setAcquireFailure(fail bool) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.failAcquire = fail
}

func (m *mockConsulLock) setSessionCreateFailure(fail bool) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.failSession = fail
}

func (m *mockConsulLock) simulateLockRelease() {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.lockHolder = ""
	m.session = ""
}

// 实现 ConsulLock 接口
func (m *mockConsulLock) CreateSession(_ *api.SessionEntry) (string, error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.failSession {
		return "", errors.New("mock session creation failed")
	}
	newSession := "mock-session-" + time.Now().String()
	m.session = newSession
	return newSession, nil
}

func (m *mockConsulLock) DestroySession(sessionID string) error {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.session == sessionID {
		m.session = ""
	}
	return nil
}

func (m *mockConsulLock) RenewSessionPeriodic(sessionID string, doneCh <-chan struct{}) error {
	<-doneCh // 模拟阻塞直到被取消
	return nil
}

func (m *mockConsulLock) AcquireLock(p *api.KVPair) (bool, error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.failAcquire {
		return false, errors.New("mock acquire failed")
	}
	if m.lockHolder == "" {
		m.lockHolder = string(p.Value)
		m.session = p.Session
		return true, nil
	}
	if m.session == p.Session {
		return true, nil // 已经持有
	}
	return false, nil // 被别人持有
}

func (m *mockConsulLock) ReleaseLock(p *api.KVPair) (bool, error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.session == p.Session {
		m.lockHolder = ""
		m.session = ""
		return true, nil
	}
	return false, nil
}

注意：为保持测试代码的清晰，mockConsulLock的MonitorLock没有实现，因为重构后的Aggregator内部循环已经通过Get轮询来模拟监控，我们可以通过控制Get的模拟行为来测试。在一个更完整的测试套件中，我们会模拟所有接口方法。

现在，我们可以编写测试用例了。

测试用例1：成功成为领导者

// aggregator_test.go (续)
func TestAggregator_BecomesLeaderSuccessfully(t *testing.T) {
	mockLock := newMockConsulLock()
	agg, err := NewAggregatorWithLock("node-1", "service/aggregator/leader", mockLock)
	assert.NoError(t, err)

	var taskExecuted bool
	var taskMu sync.Mutex
	agg.aggregationTask = func(ctx context.Context) {
		taskMu.Lock()
		taskExecuted = true
		taskMu.Unlock()
	}

	agg.Start()

	// 等待一段时间让选举循环运行
	time.Sleep(100 * time.Millisecond)

	assert.True(t, agg.IsLeader(), "Aggregator should become the leader")
	
	taskMu.Lock()
	assert.True(t, taskExecuted, "Aggregation task should have been executed")
	taskMu.Unlock()

	agg.Stop()
}

测试用例2：因锁被占用而成为追随者

// aggregator_test.go (续)
func TestAggregator_BecomesFollowerIfLockIsTaken(t *testing.T) {
	mockLock := newMockConsulLock()
	// 预先设置锁被 "other-node" 持有
	mockLock.lockHolder = "other-node"
	mockLock.session = "other-session"

	agg, err := NewAggregatorWithLock("node-1", "service/aggregator/leader", mockLock)
	assert.NoError(t, err)

	var taskExecuted bool
	agg.aggregationTask = func(ctx context.Context) {
		taskExecuted = true
	}

	agg.Start()

	time.Sleep(100 * time.Millisecond)

	assert.False(t, agg.IsLeader(), "Aggregator should be a follower")
	assert.False(t, taskExecuted, "Aggregation task should not be executed")

	agg.Stop()
}

测试用例3：成为领导者后，因锁丢失而退位

这是最关键的测试场景，模拟了领导者节点的“非自愿”退位。

// aggregator_test.go (续)
func TestAggregator_StepsDownWhenLockIsLost(t *testing.T) {
	mockLock := newMockConsulLock()
	// 为了复现这个场景，我们需要对 Aggregator 内部的 Get 调用进行模拟。
	// 这意味着我们的 ConsulLock 接口需要增加一个 Get 方法。
	// 这里为了简化，我们通过模拟锁的外部释放来触发状态变化。
	// 假设 `runAsLeader` 的监控循环能检测到这一点。

	agg, err := NewAggregatorWithLock("node-1", "service/aggregator/leader", mockLock)
	assert.NoError(t, err)

	agg.Start()

	// 1. 确认成为领导者
	time.Sleep(100 * time.Millisecond) // 等待获取锁
	assert.True(t, agg.IsLeader(), "Should initially become leader")

	// 2. 从外部模拟锁被释放（例如，session TTL过期）
	// 在真实的实现中，runAsLeader 内部的 Get 查询会发现 kv.Session 不再是自己的
	// 在我们的mock中，我们需要一种方式让 `acquireLock` 在下一轮循环失败
	mockLock.simulateLockRelease()
	mockLock.lockHolder = "another-node" // 锁被别人抢占了

	// 这里的测试需要更精细的控制。一个更好的 mock 会返回 channel，让测试代码可以推送事件。
	// 简单起见，我们假设 aggregator 的监控循环会发现锁持有者变化并退位
	// 在我们的代码中，runAsLeader 会因为 Get 发现 session 不匹配而退出，
	// 主循环会重新进入 acquireLock 阶段，此时会失败。
	
	// 等待aggregator的循环检测到变化并退位
	// 使用 assert.Eventually 来轮询状态变化
	assert.Eventually(t, func() bool {
		return !agg.IsLeader()
	}, 2*time.Second, 100*time.Millisecond, "Aggregator should step down after losing the lock")

	agg.Stop()
}

方案局限性与未来展望

通过接口抽象和依赖注入，我们成功地为一个复杂的分布式组件构建了快速、可靠的单元测试套件。这使得我们能够在不依赖外部环境的情况下，验证其核心状态转换逻辑的正确性。

然而，当前方案并非银弹。单元测试无法覆盖所有问题：

配置错误：如错误的Consul地址、错误的ACL Token等，这些只能在集成测试或运行时发现。
网络真实性：单元测试无法模拟真实世界的网络延迟、丢包或分区。DefaultSessionTTL等超时参数的合理性，仍需在更真实的环境下进行压测和混沌工程实验来验证。
Consul行为差异：我们的mockConsulLock模拟了我们理解的Consul行为。如果Consul在未来版本中改变了某些边缘情况下的行为，我们的模拟可能会与现实脱节。因此，保留一套覆盖关键路径的、运行频率较低的集成测试仍然是必要的。

未来的优化路径可以包括：

Split-Brain防护：当前的TTL机制能有效应对节点宕机，但在网络分区场景下，被隔离的旧领导者可能在短时间内（直到其Session过期）无法感知自己已不再是领导者。对于需要强一致性的聚合任务，可能需要引入Fencing Token机制，确保TSDB只接受来自当前合法领导者的数据。
更精细的监控：除了节点是否为领导者这个二进制状态外，还应暴露更多内部指标，如Session续期成功/失败次数、锁获取尝试次数、成为领导者/追随者的时长等，并接入时序监控系统，以便更早地发现潜在问题。

Consul Go 时序单元测试高可用

使用 Jenkins 与 Azure Service Bus 编排跨云 Serverless 构建工作流

2023-10-27 DevOps

Jenkins Serverless AWS Azure Service Bus 构建与工具

基于NumPy与Redis分布式锁构建高性能的分布式滑动窗口计数器

2023-10-27 后端架构

Spring Boot NumPy 分布式锁 Redis 高性能