共计 3147 个字符，预计需要花费 8 分钟才能阅读完成。
背景痛点
在分布式系统中，传统的任务调度方式如 crontab 和消息队列存在明显的局限性。crontab 虽然简单易用，但缺乏分布式支持，容易出现单点故障。消息队列虽然能够实现分布式调度，但在高负载情况下容易引发雪崩效应，导致系统整体性能下降。
- 单点故障 ：传统的 crontab 依赖于单个节点的定时任务，一旦该节点宕机，整个调度系统将失效。
- 雪崩效应 ：消息队列在高并发场景下，任务堆积可能导致系统资源耗尽，进而引发连锁反应。
- 缺乏弹性 ：传统方案难以动态调整任务分配，无法应对突发流量或节点故障。
架构设计
协调服务选型
在分布式系统中，协调服务的选择至关重要。我们对比了 ZooKeeper 和 etcd 两种主流方案：
- ZooKeeper：
- 优点：成熟稳定，支持强一致性。
- 缺点：写性能较差，配置复杂。
- etcd：
- 优点：轻量级，高可用，支持 gRPC 接口。
- 缺点：社区支持相对较少。
最终我们选择了 etcd，因其更适合高并发场景下的动态配置管理。
Agent 集群架构
采用 Master-Worker 模式，架构图如下：
@startuml
node "Master" {
component "API Server"
component "Scheduler"
component "etcd Client"
}
node "Worker 1" {
component "Agent"
component "Task Executor"
}
node "Worker 2" {
component "Agent"
component "Task Executor"
}
"API Server" --> "etcd Client"
"etcd Client" --> "Scheduler"
"Scheduler" --> "Worker 1" : gRPC
"Scheduler" --> "Worker 2" : gRPC
@enduml核心实现
任务分片算法
使用一致性哈希算法实现任务分片，确保任务均匀分布且节点变动时影响最小。以下是 Go 语言实现示例：
// 一致性哈希环实现
type HashRing struct {nodes []string
hashFunc func(string) uint32
replicas int
ring map[uint32]string
sortedKeys []uint32}
func NewHashRing(replicas int, fn func(string) uint32) *HashRing {
return &HashRing{
replicas: replicas,
hashFunc: fn,
ring: make(map[uint32]string),
}
}
// 添加节点
func (h *HashRing) Add(nodes ...string) {
for _, node := range nodes {
for i := 0; i < h.replicas; i++ {hash := h.hashFunc(fmt.Sprintf("%s:%d", node, i))
h.ring[hash] = node
h.sortedKeys = append(h.sortedKeys, hash)
}
}
sort.Slice(h.sortedKeys, func(i, j int) bool {return h.sortedKeys[i] < h.sortedKeys[j]
})
}
// 获取任务对应的节点
func (h *HashRing) Get(key string) string {if len(h.ring) == 0 {return ""}
hash := h.hashFunc(key)
idx := sort.Search(len(h.sortedKeys), func(i int) bool {return h.sortedKeys[i] >= hash
})
if idx == len(h.sortedKeys) {idx = 0}
return h.ring[h.sortedKeys[idx]]
}gRPC 长连接保活机制
为保证 Agent 与 Master 之间的稳定通信，实现了基于 gRPC 的心跳检测机制：
// 心跳检测服务端实现
type HealthServer struct {
pb.UnimplementedHealthServer
timeout time.Duration
}
func (s *HealthServer) Check(ctx context.Context, req *pb.HealthCheckRequest) (*pb.HealthCheckResponse, error) {
select {case <-ctx.Done():
return nil, status.Error(codes.Canceled, "client cancelled")
default:
if time.Since(req.Timestamp.AsTime()) > s.timeout {return nil, status.Error(codes.DeadlineExceeded, "heartbeat timeout")
}
return &pb.HealthCheckResponse{Status: pb.HealthCheckResponse_SERVING}, nil
}
}
// 客户端心跳发送
type HeartbeatClient struct {
client pb.HealthClient
stream pb.Health_CheckClient
cancel context.CancelFunc
}
func (c *HeartbeatClient) Start() error {ctx, cancel := context.WithCancel(context.Background())
c.cancel = cancel
stream, err := c.client.Check(ctx)
if err != nil {return err}
go func() {ticker := time.NewTicker(5 * time.Second)
defer ticker.Stop()
for {
select {
case <-ticker.C:
if err := stream.Send(&pb.HealthCheckRequest{Timestamp: timestamppb.Now(),
}); err != nil {log.Printf("send heartbeat failed: %v", err)
return
}
case <-ctx.Done():
return
}
}
}()
return nil
}生产考量
压测数据
通过压力测试，我们得出了 QPS 与 Worker 数量的关系曲线：
- 当 Worker 数量从 1 增加到 10 时，QPS 线性增长
- 超过 20 个 Worker 后，由于网络开销增加，QPS 增长趋缓
- 最终系统吞吐量提升了 300%
磁盘写放大问题
为避免频繁的小文件写入导致磁盘性能下降，采取了以下措施：
- 批量写入：将多个任务状态变更合并为一个批量操作
- 内存缓存：使用 Redis 缓存热数据，减少磁盘 IO
- 压缩存储：对历史任务日志进行压缩存储
避坑指南
Agent 版本灰度升级
采用分批次升级策略：
- 先升级 5% 的 Agent 节点
- 观察 24 小时无异常后，逐步扩大升级范围
- 最终完成 100% 升级
任务积压动态降级
当系统检测到任务积压时，自动触发降级策略：
@startuml
state "正常模式" as normal
state "降级模式" as degraded
[*] --> normal
normal --> degraded : 任务积压超过阈值
degraded --> normal : 积压缓解
@enduml具体降级方法包括：
- 跳过非关键任务
- 延长任务执行间隔
- 限制新任务接收速率
开放性问题
在实际生产环境中，跨地域调度是一个复杂的问题。如何实现高效的跨地域任务调度？以下是几个可能的思考方向：
- 基于地理位置的负载均衡
- 多中心数据同步策略
- 网络延迟优化方案
欢迎在评论区分享你的见解和经验。
正文完
