elk_coordinator

README

ELK Coordinator

ELK Coordinator 是一个高性能、可扩展的分布式任务处理框架，专为需要大规模并行处理的系统设计。它采用Leader-Worker架构，通过分区机制将工作负载分配给多个工作节点，支持动态扩缩容和故障自动恢复。

特性

核心功能

• 自动Leader选举: 基于分布式锁的自动Leader选举机制，无需手动指定主节点
• 工作节点自动发现: 通过心跳机制自动发现和管理工作节点
• 分区任务处理: 将工作负载分割为可管理的分区，支持并行处理
• 任务窗口: 单节点可并行处理多个分区，提高资源利用率
• 分区抢占: 支持工作节点抢占其他节点的分区，提高资源分配效率

分区策略

• SimpleStrategy: 基于分布式锁的简单分区策略，适用于中小规模场景
• HashPartitionStrategy: 基于Redis Hash的高性能分区策略，支持更高并发和更细粒度的控制

可靠性保障

• 熔断保护: 内置熔断器机制，防止系统过载并提高稳定性
• 故障自动恢复: 自动检测和处理节点故障，重新分配失败的分区
• 优雅启停: 支持平滑的启动和关闭流程，避免任务丢失

监控与可观测性

• Prometheus 指标系统: 完整的指标收集系统，支持 Leader 状态、任务处理、心跳等30+项指标
• Grafana 仪表板: 预配置的可视化仪表板，实时展示系统运行状态
• 实时监控: 支持 HTTP 端点暴露指标，可与 Prometheus 无缝集成
• 多维度指标: Leader 视角、Worker 视角、分区处理、错误统计等全面监控
• 一键启动监控栈: 提供 Docker Compose 配置，快速搭建完整监控环境
• Gap检测: 智能检测和处理数据处理中的间隙

灵活性

• 可扩展接口: 提供灵活的接口允许自定义处理逻辑和分区规划
• 丰富的配置选项: 支持细粒度的系统配置和调优
• 模块化设计: 各组件职责分离，便于测试和维护

安装

go get github.com/iheCoder/elk_coordinator

快速开始

以下是一个完整的示例，展示如何使用ELK Coordinator处理任务：

package main

import (
    "context"
    "fmt"
    "time"
    
    "github.com/redis/go-redis/v9"
    "github.com/yourusername/elk_coordinator"
    "github.com/yourusername/elk_coordinator/data"
    "github.com/yourusername/elk_coordinator/leader"
    "github.com/yourusername/elk_coordinator/model"
)

// 自定义任务处理器实现
type MyProcessor struct{}

// Process 处理指定ID范围内的任务
func (p *MyProcessor) Process(ctx context.Context, minID int64, maxID int64, options map[string]interface{}) (int64, error) {
    fmt.Printf("处理ID范围: %d - %d\n", minID, maxID)
    
    // 这里实现你的任务处理逻辑
    // 例如：处理数据库中的一批记录、处理队列中的消息等
    // time.Sleep(100 * time.Millisecond) // 模拟处理时间
    
    processedCount := maxID - minID + 1
    return processedCount, nil
}

// 自定义分区规划器实现
type MyPartitionPlaner struct{}

// PartitionSize 返回建议的分区大小
func (p *MyPartitionPlaner) PartitionSize(ctx context.Context) (int64, error) {
    return 1000, nil // 每个分区处理1000条记录
}

// GetNextMaxID 获取下一批次的最大ID
func (p *MyPartitionPlaner) GetNextMaxID(ctx context.Context, startID int64, rangeSize int64) (int64, error) {
    // 这里可以从数据库或其他数据源获取实际的ID上限
    // 简单示例：每次增加rangeSize
    return startID + rangeSize - 1, nil
}

func main() {
    // 创建Redis客户端
    redisClient := redis.NewClient(&redis.Options{
        Addr: "localhost:6379",
    })
    
    // 创建数据存储
    dataStore := data.NewRedisDataStore(redisClient, data.DefaultOptions())
    
    // 创建处理器和分区规划器
    processor := &MyProcessor{}
    planer := &MyPartitionPlaner{}
    
    // 创建管理器，使用Hash分区策略
    mgr := elk_coordinator.NewMgr(
        "my-app",                              // 应用命名空间，用于隔离不同应用的数据
        dataStore,                             // 数据存储
        processor,                             // 任务处理器
        planer,                                // 分区规划器
        model.StrategyTypeHash,                // 分区策略类型：Hash策略
        elk_coordinator.WithTaskWindow(5),     // 启用任务窗口并设置大小（并行处理5个分区）
        elk_coordinator.WithHeartbeatInterval(5*time.Second), // 自定义心跳间隔
        elk_coordinator.WithAllowPreemption(true), // 允许分区抢占
    )
    
    // 创建上下文（可以通过取消此上下文来停止服务）
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    
    // 启动协调器
    if err := mgr.Start(ctx); err != nil {
        fmt.Printf("启动失败: %v\n", err)
        return
    }
    
    fmt.Println("服务已启动，节点ID:", mgr.ID)
    fmt.Println("按Ctrl+C停止...")
    
    // 等待终止信号
    <-ctx.Done()
    
    // 停止服务
    mgr.Stop()
    fmt.Println("服务已停止")
}

核心概念

分区 (Partition)

分区是任务处理的基本单位，由ID范围（最小ID和最大ID）定义。系统将工作负载分割为多个分区，并分配给不同的工作节点处理。

分区状态包括：

• pending: 等待被处理
• claimed: 已被工作节点认领
• running: 正在处理中
• completed: 处理完成
• failed: 处理失败

分区策略

系统支持两种分区策略：

SimpleStrategy (简单策略)

• 基于分布式锁实现
• 适用于中小规模场景
• 使用Redis String类型存储分区信息
• 策略类型：model.StrategyTypeSimple

HashPartitionStrategy (哈希策略)

• 基于Redis Hash实现，性能更高
• 支持更高并发和更细粒度的控制
• 提供更丰富的统计和查询功能
• 策略类型：model.StrategyTypeHash（推荐）

Leader选举

系统使用分布式锁实现自动Leader选举。Leader节点负责：

• 监控活跃工作节点
• 创建和分配分区
• 处理节点故障和任务重新分配
• 执行系统级的协调任务

任务窗口

任务窗口允许单个工作节点同时处理多个分区，提高资源利用率。窗口大小决定了并发处理的分区数量。特性：

• 可配置窗口大小
• 支持动态调整
• 自动负载均衡

分区抢占

分区抢占机制允许工作节点主动获取其他节点持有但可能已失效的分区：

• 提高资源分配效率
• 减少因节点故障导致的任务延迟
• 可通过配置选项启用/禁用

熔断器

内置的熔断器机制可以检测连续失败，并在系统压力过大时暂时阻止新任务的处理，防止系统崩溃。支持：

• 连续失败阈值检测
• 自动恢复机制
• Half-Open状态探测
• 失败统计和分析

配置选项

ELK Coordinator提供了丰富的配置选项，可以通过选项模式灵活配置：

mgr := elk_coordinator.NewMgr(
    "my-namespace",                                        // 应用命名空间
    dataStore,                                            // 数据存储实例
    processor,                                            // 任务处理器
    planer,                                              // 分区规划器
    model.StrategyTypeHash,                              // 分区策略类型
    
    // 基础配置
    elk_coordinator.WithLogger(customLogger),            // 自定义日志记录器
    elk_coordinator.WithHeartbeatInterval(10*time.Second), // 心跳间隔
    elk_coordinator.WithLeaderElectionInterval(5*time.Second), // Leader选举间隔
    
    // 锁配置
    elk_coordinator.WithPartitionLockExpiry(3*time.Minute), // 分区锁过期时间
    elk_coordinator.WithLeaderLockExpiry(30*time.Second),   // Leader锁过期时间
    
    // 分区配置
    elk_coordinator.WithWorkerPartitionMultiple(3),      // 工作节点分区倍数
    elk_coordinator.WithTaskWindow(5),                   // 任务窗口大小
    elk_coordinator.WithAllowPreemption(true),           // 启用分区抢占
    elk_coordinator.WithAllocationInterval(2*time.Minute), // 分区分配检查间隔
)

主要配置参数

参数	类型	默认值	说明
HeartbeatInterval	time.Duration	10秒	工作节点心跳间隔
LeaderElectionInterval	time.Duration	180秒	Leader选举检查间隔
PartitionLockExpiry	time.Duration	3分钟	分区锁的过期时间
LeaderLockExpiry	time.Duration	30秒	Leader锁的过期时间
WorkerPartitionMultiple	int64	5	每个工作节点分配的分区倍数
TaskWindowSize	int	1	任务窗口大小（并行处理的分区数）
AllowPreemption	bool	false	是否允许抢占其他节点的分区
AllocationInterval	time.Duration	2分钟	分区分配检查间隔

核心接口

Processor 接口

任务处理器接口，定义如何处理分区内的任务：

type Processor interface {
    // Process 处理指定ID范围内的任务
    // 参数:
    // - ctx: 处理上下文，可用于取消或设置超时
    // - minID: 处理范围的最小ID（包含）
    // - maxID: 处理范围的最大ID（包含）  
    // - options: 处理选项，可以包含任何特定于任务的参数
    // 返回:
    // - 处理的项目数量
    // - 处理过程中遇到的错误
    Process(ctx context.Context, minID int64, maxID int64, options map[string]interface{}) (int64, error)
}

PartitionPlaner 接口

分区规划器接口，定义如何规划分区大小和ID范围：

type PartitionPlaner interface {
    // PartitionSize 返回建议的分区大小
    PartitionSize(ctx context.Context) (int64, error)
    
    // GetNextMaxID 获取下一批次的最大ID
    // 参数:
    // - startID: 起始ID
    // - rangeSize: 范围大小
    // 返回下一批次处理的最大ID
    GetNextMaxID(ctx context.Context, startID int64, rangeSize int64) (int64, error)
}

PartitionStrategy 接口

分区策略接口，定义分区管理的核心操作：

type PartitionStrategy interface {
    // 基础CRUD操作
    GetPartition(ctx context.Context, partitionID int) (*model.PartitionInfo, error)
    GetAllPartitions(ctx context.Context) ([]*model.PartitionInfo, error)
    DeletePartition(ctx context.Context, partitionID int) error
    GetFilteredPartitions(ctx context.Context, filters model.GetPartitionsFilters) ([]*model.PartitionInfo, error)
    
    // 批量操作
    CreatePartitionsIfNotExist(ctx context.Context, request model.CreatePartitionsRequest) ([]*model.PartitionInfo, error)
    DeletePartitions(ctx context.Context, partitionIDs []int) error
    
    // 并发安全操作
    UpdatePartition(ctx context.Context, partitionInfo *model.PartitionInfo, options *model.UpdateOptions) (*model.PartitionInfo, error)
    
    // 分布式协调
    AcquirePartition(ctx context.Context, partitionID int, nodeID string, options *model.AcquireOptions) (*model.PartitionInfo, error)
    ReleasePartition(ctx context.Context, partitionID int, nodeID string) error
    
    // 策略信息
    StrategyType() model.StrategyType
}

DataStore 接口

数据存储接口，定义底层存储操作（默认由RedisDataStore实现）：

type DataStore interface {
    // 分布式锁操作
    AcquireLock(ctx context.Context, key string, value string, expiry time.Duration) (bool, error)
    RenewLock(ctx context.Context, key string, value string, expiry time.Duration) (bool, error)
    ReleaseLock(ctx context.Context, key string, value string) error
    
    // 心跳和工作节点管理
    SetHeartbeat(ctx context.Context, key string, value string) error
    RegisterWorker(ctx context.Context, workersKey, workerID string, heartbeatKey string, heartbeatValue string) error
    UnregisterWorker(ctx context.Context, workersKey, workerID string, heartbeatKey string) error
    
    // 分区和状态操作
    SetPartitions(ctx context.Context, key string, value string) error
    GetPartitions(ctx context.Context, key string) (string, error)
    // ... 其他方法
}

高级功能

熔断器配置

可以自定义熔断器行为来保护系统免受过载影响：

// 注意：熔断器功能目前在task包中定义，但还未完全集成到主管理器中
// 以下为熔断器配置的示例代码结构

circuitBreakerConfig := &task.CircuitBreakerConfig{
    ConsecutiveFailureThreshold: 5,              // 连续失败触发熔断阈值
    TotalFailureThreshold:       10,             // 失败分区数阈值
    OpenTimeout:                 30*time.Second, // 熔断开启后等待恢复的时间
    MaxHalfOpenRequests:         2,              // Half-Open状态下最大探测请求数
    FailureTimeWindow:           5*time.Minute,  // 失败统计时间窗口
}

// 熔断器状态包括：
// - "closed": 正常状态，允许所有请求通过
// - "open": 熔断状态，拒绝新的请求
// - "halfopen": 半开状态，允许少量探测请求

指标收集和监控

系统自动收集多种性能指标，用于监控和调优：

// 获取分区统计信息（以Hash策略为例）
partitionStats, err := hashStrategy.GetPartitionStats(ctx)
if err == nil {
    fmt.Printf("总分区数: %d\n", partitionStats.TotalCount)
    fmt.Printf("已完成分区数: %d\n", partitionStats.CompletedCount)
    fmt.Printf("运行中分区数: %d\n", partitionStats.RunningCount)
    fmt.Printf("失败分区数: %d\n", partitionStats.FailedCount)
}

分区查询和过滤

支持丰富的分区查询功能：

// 查询特定状态的分区
filters := model.GetPartitionsFilters{
    Status: []string{"running", "failed"},
    NodeID: "specific-node-id",
}

partitions, err := strategy.GetFilteredPartitions(ctx, filters)
if err == nil {
    for _, partition := range partitions {
        fmt.Printf("分区 %d: 状态=%s, 节点=%s\n", 
            partition.PartitionID, partition.Status, partition.NodeID)
    }
}

Gap检测功能

系统提供智能的数据处理间隙检测功能：

// Gap检测能够：
// - 自动识别数据处理中的间隙
// - 提供增量和离散模式的Gap检测
// - 支持自定义Gap检测策略
// - 详细的Gap分析和报告
// 
// 具体使用方法请参考 examples/ 目录下的示例代码

批量操作

支持高效的批量分区操作：

// 批量创建分区
createRequest := model.CreatePartitionsRequest{
    Partitions: []*model.PartitionInfo{
        {PartitionID: 1, MinID: 1, MaxID: 1000, Status: "pending"},
        {PartitionID: 2, MinID: 1001, MaxID: 2000, Status: "pending"},
        // ... 更多分区
    },
}

createdPartitions, err := strategy.CreatePartitionsIfNotExist(ctx, createRequest)

// 批量删除分区
err = strategy.DeletePartitions(ctx, []int{1, 2, 3})

性能调优建议

1. 选择合适的分区策略：

   • 小规模场景使用SimpleStrategy
   • 大规模高并发场景使用HashPartitionStrategy

2. 合理配置任务窗口大小：

   • 根据系统资源和任务复杂度调整
   • 过大可能导致资源争用，过小影响吞吐量

3. 优化心跳和锁过期时间：

   • 心跳间隔不宜过短，避免网络开销
   • 锁过期时间要考虑任务处理时间

4. 监控系统指标：

   • 定期检查分区状态分布和处理速率
   • 监控失败率和处理速度趋势
   • 关注节点健康状态和心跳情况
   • 设置合理的告警阈值，及时发现问题

架构设计

整体架构

┌─────────────────────────────────────────────────────────────┐
│                    ELK Coordinator                          │
├─────────────────────────────────────────────────────────────┤
│  Manager (mgr.go)                                          │
│  ├── Leader Management     ├── Task Processing             │
│  │   ├── Election         │   ├── Task Window              │
│  │   ├── Worker Discovery │   ├── Circuit Breaker          │
│  │   └── Partition Assignment │   └── Task Runner         │
│  └── Partition Strategy    └── Data Storage                │
│      ├── SimpleStrategy    │   └── Redis DataStore         │
│      └── HashPartitionStrategy                             │
└─────────────────────────────────────────────────────────────┘

核心组件

1. 管理器 (Manager)：协调所有组件，处理启动/停止流程
2. Leader管理：负责选举和工作节点发现
3. 分区策略：管理分区的创建、分配和状态更新
4. 任务处理：执行具体的任务处理逻辑
5. 数据存储：提供持久化和分布式协调

工作流程

1. 初始化阶段：

   • 节点启动并注册到集群
   • 选举Leader节点
   • 初始化分区策略

2. 分区创建：

   • Leader根据PartitionPlaner创建分区
   • 分区信息存储到数据存储中

3. 任务分配：

   • Leader监控活跃工作节点
   • 根据负载均衡策略分配分区
   • 工作节点获取并锁定分区

4. 任务执行：

   • 工作节点处理分配的分区
   • 更新分区状态和进度
   • 处理完成后释放分区

5. 故障处理：

   • 检测节点故障和分区超时
   • 重新分配失败的分区
   • 触发熔断器保护机制

示例和最佳实践

数据库批处理示例

// 处理数据库记录的示例
type DatabaseProcessor struct {
    db *sql.DB
}

func (p *DatabaseProcessor) Process(ctx context.Context, minID int64, maxID int64, options map[string]interface{}) (int64, error) {
    query := "UPDATE records SET processed = true WHERE id BETWEEN ? AND ?"
    result, err := p.db.ExecContext(ctx, query, minID, maxID)
    if err != nil {
        return 0, err
    }
    
    rowsAffected, err := result.RowsAffected()
    return rowsAffected, err
}

type DatabasePartitionPlaner struct {
    db *sql.DB
}

func (p *DatabasePartitionPlaner) PartitionSize(ctx context.Context) (int64, error) {
    return 1000, nil // 每次处理1000条记录
}

func (p *DatabasePartitionPlaner) GetNextMaxID(ctx context.Context, startID int64, rangeSize int64) (int64, error) {
    var maxID int64
    query := "SELECT COALESCE(MAX(id), ?) FROM records WHERE id >= ?"
    err := p.db.QueryRowContext(ctx, query, startID+rangeSize-1, startID).Scan(&maxID)
    
    // 确保不超过实际的ID范围
    if maxID < startID+rangeSize-1 {
        return maxID, nil
    }
    return startID + rangeSize - 1, err
}

消息队列处理示例

// 处理消息队列的示例
type MessageQueueProcessor struct {
    queue MessageQueue
}

func (p *MessageQueueProcessor) Process(ctx context.Context, minID int64, maxID int64, options map[string]interface{}) (int64, error) {
    processed := int64(0)
    
    for i := minID; i <= maxID; i++ {
        select {
        case <-ctx.Done():
            return processed, ctx.Err()
        default:
            message, err := p.queue.GetMessage(i)
            if err != nil {
                continue // 跳过错误消息
            }
            
            if err := p.processMessage(message); err != nil {
                return processed, err
            }
            processed++
        }
    }
    
    return processed, nil
}

测试

项目包含完整的测试套件：

# 运行所有测试
go test ./...

# 运行特定包的测试
go test ./partition
go test ./task
go test ./leader

# 运行集成测试
go test -tags=integration ./...

# 运行性能测试
go test -bench=. ./...

故障排查

常见问题

1. Redis连接失败

   # 检查Redis是否运行
   redis-cli ping
   

2. 分区锁定时间过长

   • 检查 PartitionLockExpiry 配置
   • 确认任务处理时间是否合理

3. Leader选举频繁

   • 检查网络连接稳定性
   • 调整 LeaderLockExpiry 和 HeartbeatInterval

4. 分区处理失败率高

   • 检查Processor实现的错误处理
   • 考虑启用熔断器保护

性能基准

在标准测试环境下的性能数据：

• 吞吐量: 10,000+ 分区/秒 (HashPartitionStrategy)
• 延迟: < 10ms 分区获取时间
• 并发: 支持100+ 工作节点
• 可靠性: 99.9%+ 分区处理成功率

路线图

• 完全集成熔断器功能到主管理器
• 支持更多数据存储后端 (etcd, Consul)
• 完整的监控和可观测性系统
  • Prometheus 指标收集
  • Grafana 仪表板
  • Docker Compose 一键部署
  • 30+ 项核心指标
• 增加Web管理界面
• 支持分区优先级
• 增加告警规则模板和通知集成
• 支持动态配置更新

许可证

本项目使用 MIT 许可证 - 查看 LICENSE 文件了解详细信息。

Documentation

Index

• type Mgr
  • func NewMgr(namespace string, dataStore data.DataStore, processor task.Processor, ...) *Mgr
  • func (m *Mgr) Handle(ctx context.Context) error
  • func (m *Mgr) Lead(ctx context.Context) error
  • func (m *Mgr) Start(ctx context.Context) error
  • func (m *Mgr) Stop()
• type MgrOption
  • func WithAllocationInterval(interval time.Duration) MgrOption
  • func WithAllowPreemption(allowPreemption bool) MgrOption
  • func WithHeartbeatInterval(interval time.Duration) MgrOption
  • func WithLeaderElectionInterval(interval time.Duration) MgrOption
  • func WithLeaderLockExpiry(expiry time.Duration) MgrOption
  • func WithLogger(logger utils.Logger) MgrOption
  • func WithMetricsAddr(addr string) MgrOption
  • func WithMetricsEnabled(enabled bool) MgrOption
  • func WithPartitionLockExpiry(expiry time.Duration) MgrOption
  • func WithTaskWindow(windowSize int) MgrOption
  • func WithWorkerPartitionMultiple(multiple int64) MgrOption

Types

type Mgr

type Mgr struct {
	// 核心字段
	ID              string
	Namespace       string
	DataStore       data.DataStore
	TaskProcessor   task.Processor
	PartitionPlaner leader.PartitionPlaner
	Logger          utils.Logger

	// 分区策略配置
	PartitionStrategyType model.StrategyType          // 分区策略类型
	PartitionStrategy     partition.PartitionStrategy // 分区策略实例

	// 配置选项
	HeartbeatInterval       time.Duration // 心跳间隔
	LeaderElectionInterval  time.Duration // Leader选举间隔
	PartitionLockExpiry     time.Duration // 分区锁过期时间
	LeaderLockExpiry        time.Duration // Leader锁过期时间
	WorkerPartitionMultiple int64         // 每个工作节点分配的分区倍数，用于计算ID探测范围
	AllocationInterval      time.Duration // 分区分配检查间隔，默认2分钟
	AllowPreemption         bool          // 是否允许抢占其他节点的分区，默认false

	// 任务窗口配置
	TaskWindowSize int // 任务窗口大小（同时处理的最大分区数）

	// 监控配置
	MetricsEnabled bool                    // 是否启用监控系统
	MetricsAddr    string                  // 监控服务地址
	MetricsManager *metrics.MetricsManager // 监控管理器
	// contains filtered or unexported fields
}

Mgr 是一个分布式管理器，管理分区任务的执行

func NewMgr

func NewMgr(namespace string, dataStore data.DataStore, processor task.Processor, planer leader.PartitionPlaner, strategyType model.StrategyType, options ...MgrOption) *Mgr

NewMgr 创建一个新的管理器实例

func (*Mgr) Handle

func (m *Mgr) Handle(ctx context.Context) error

Handle 处理分区任务的主循环

func (*Mgr) Lead

func (m *Mgr) Lead(ctx context.Context) error

Lead 处理Leader相关的工作

func (*Mgr) Start

func (m *Mgr) Start(ctx context.Context) error

Start 启动管理器

func (*Mgr) Stop

func (m *Mgr) Stop()

Stop 停止管理器，协调各组件的优雅关闭

type MgrOption

type MgrOption func(*Mgr)

MgrOption 定义管理器的配置选项

func WithAllocationInterval

func WithAllocationInterval(interval time.Duration) MgrOption

WithAllocationInterval 设置分区分配检查间隔

func WithAllowPreemption

func WithAllowPreemption(allowPreemption bool) MgrOption

WithAllowPreemption 设置是否允许抢占其他节点的分区

func WithHeartbeatInterval

func WithHeartbeatInterval(interval time.Duration) MgrOption

WithHeartbeatInterval 设置心跳间隔

func WithLeaderElectionInterval

func WithLeaderElectionInterval(interval time.Duration) MgrOption

WithLeaderElectionInterval 设置Leader选举间隔

func WithLeaderLockExpiry

func WithLeaderLockExpiry(expiry time.Duration) MgrOption

WithLeaderLockExpiry 设置Leader锁过期时间

func WithLogger

func WithLogger(logger utils.Logger) MgrOption

WithLogger 设置自定义日志记录器

func WithMetricsAddr

func WithMetricsAddr(addr string) MgrOption

WithMetricsAddr 设置监控服务地址

func WithMetricsEnabled

func WithMetricsEnabled(enabled bool) MgrOption

WithMetricsEnabled 设置是否启用监控系统

func WithPartitionLockExpiry

func WithPartitionLockExpiry(expiry time.Duration) MgrOption

WithPartitionLockExpiry 设置分区锁过期时间

func WithTaskWindow

func WithTaskWindow(windowSize int) MgrOption

WithTaskWindow 设置任务窗口大小

func WithWorkerPartitionMultiple

func WithWorkerPartitionMultiple(multiple int64) MgrOption

WithWorkerPartitionMultiple 设置工作节点分区倍数