cydist

package
v1.1.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 26, 2026 License: MIT Imports: 22 Imported by: 0

Documentation

Overview

distlock.go

Package cydist provides a generic in-memory event store backed by Redis.

Package cydist 提供带主从选举与任务队列的分布式调度器

项目用法总览：

  1. 快速开始： s := cydist.NewScheduler(redisClient, cydist.WithSchedulerNamespace("default")) s.RegisterFactory("my-factory", func(taskID string, metadata json.RawMessage) (func(context.Context) error, error) { return func(ctx context.Context) error { return nil }, nil }) td := cydist.NewDailyTask("daily-job", "my-factory", []cydist.AtTime{cydist.NewAtTime(9,0,0)}, cydist.WithDefTimeout(30*time.Second)) s.AddTask(td) s.Start()

  2. 添加任务时同时注册工厂（无需手动 key）： // 优先使用 TaskDefinition.FactoryKey；为空时自动回退为任务 ID s.AddTask( cydist.NewEveryTask("clean", "", time.Hour), cydist.WithFactory(func(taskID string, metadata json.RawMessage) (func(context.Context) error, error) { return func(ctx context.Context) error { return nil }, nil }), )

3) 便捷构造函数与可选项：

  • 构造：NewCronTask/NewSpecTask/NewEveryTask/NewRandomTask/NewDailyTask/NewWeeklyTask/NewMonthlyTask NewMonthlyLastDayTask/NewMonthlyLastBusinessDayTask/NewMonthlyNthWeekdayTask
  • 可选：WithDefMetadata/WithDefMetadataRaw/WithDefTimeout/WithDefStartAt/WithDefDeadline/ WithDefStartImmediately/WithDefIntervalFromCompletion

4) 启动与时间行为：

  • StartTime 与 StartImmediately 互斥：有 StartTime 时忽略 StartImmediately
  • StartTime 为过去使用 WithStartDateTimePast，从过去时间点计算下一次到期，不强制立即执行

5) 分布式与单机：

  • 提供 Redis 客户端时启用主从选举与队列广播；否则单机模式直接本地调度

支持的调度写法（Spec）： 1) Cron 表达式：

  • 5 字段：分 时 日 月 周
  • 6 字段：秒 分 时 日 月 周

2) 间隔类（从当前时刻起算）：

  • @everyNs / @everyNm / @everyNh / @everyNd
  • 允许空格：@every 2s

3) 每日定点（按系统日历）：

  • @dailyHH:MM[[:SS]]，支持多个时间：@daily08:00,20:00

4) 每周定点：

  • @weekly<day-list>/<time-list>
  • day-list 使用 0-6 表示周日到周六，支持逗号与范围（如 1,3,5 或 1-5）
  • time-list 逗号分隔多个时间（HH:MM 或 HH:MM:SS）
  • 示例：@weekly1/08:00、@weekly1,3,5/09:30、@weekly1-5/18:00、@weekly0/23:50

5) 每月定点：

  • @monthly<day-list>/<time-list>
  • day-list 支持 1-31，逗号与范围（如 1,15 或 3-12）
  • 示例：@monthly1/00:00、@monthly1,15/12:00、@monthly3,6,9,12/11:32,12:23

6) 每月最后一天：

  • @monthlyL/<time>，如 @monthlyL/23:59
  • 内部按每月 28-31 日生成 Cron，并在执行时仅保留“最后一天”
  • 支持“倒数第x天”：@monthlyLx/<time>（x>=1），例如 @monthlyL3/10:00 表示每月倒数第3天 10:00 也支持 @monthlyL-<n>/<time> 写法，如 @monthlyL-2/10:00 表示“最后一天往前第2天”（等价于倒数第3天）

7) 工作日定点：

  • @workdays/<time-list>，如 @workdays/09:00,18:00（等价每周一到五的定点执行）

8) 每月最后一个工作日：

  • @monthlyLB/<time>，如 @monthlyLB/18:00（仅在当月最后一个非周六/周日的日期执行）

9) 每月第 N 个某星期几：

  • @monthlyW<d>/<n>/<time-list>，如 @monthlyW3/2/09:00 表示每月第2个周三 09:00

说明： - 未以 @ 前缀的 Spec 视为原始 Cron 表达式 - 时间参数支持多个值，使用逗号分隔 - @every 支持 s/m/h/d 四种单位 - @daily/@weekly/@monthly 的时间支持 HH:MM 或 HH:MM:SS

使用说明：

  • 特殊 spec： Every("cache-clean", "cache-factory", 5*time.Minute) Random("random-sync", "sync-factory", 1*time.Minute, 5*time.Minute) Daily("backup", "backup-factory", []AtTime{NewAtTime(2,0,0)}) Workdays("report", "report-factory", []AtTime{NewAtTime(9,0,0)}) Weekly("sync", "sync-factory", []int{1,3,5}, []AtTime{NewAtTime(18,0,0)}) Monthly("billing", "bill-factory", []int{1,15}, []AtTime{NewAtTime(0,0,0)}) MonthlyL("archive", "archive-factory", 0, []AtTime{NewAtTime(23,59,0)}) MonthlyLB("closing", "close-factory", []AtTime{NewAtTime(18,0,0)}) MonthlyNthW("finance", "fin-factory", 3, 2, []AtTime{NewAtTime(9,0,0)}) Cron("db-backup", "backup-factory", "0 3 * * *") Spec("night-shift", "ops-factory", "@workdays/21:00") Spec("limited", "ops-factory", "[2025-01-01T00:00:00+08:00,2025-02-01T00:00:00+08:00] @every 5m") Spec("start-only", "ops-factory", "[2025-01-01 00:00:00,] @daily08:00") 说明：[start,end] 是一个闭区间写法，用于将 StartTime/Deadline 嵌入到 Spec 中； start/end 可选（[start,] / [,end]）；时间格式优先 RFC3339，其次本地格式 2006-01-02 15:04:05。
  • 可选参数（链式）： WithMeta(v), WithMetaRaw(raw), WithTimeout(d), WithStartAt(t), WithDeadline(t), StartNow(), IntervalFromCompletion()
  • 启动行为： 当同时设置 StartTime 与 StartImmediately 时，优先使用 StartTime；仅在 StartTime 为空且 StartImmediately=true 时立即启动。 StartTime 为过去时间使用 WithStartDateTimePast，从该时间点计算下一次到期；不会强制立刻执行。

Index

Constants

View Source 
const (
	DefaultLeaderTTL      = 15 * time.Second
	DefaultReelectJitter  = 3 * time.Second
	DefaultHealthInterval = 5 * time.Second // Should be LeaderTTL / 3 for safety
)
View Source 
const DefaultTaskTimeout = 30 * time.Second

Variables

This section is empty.

Functions

func CacheWrap 

func CacheWrap[T any](
	w *CacheWrapper,
	fn func(context.Context) (T, error),
	opts ...CacheFuncOption,
) func(context.Context) (T, error)

Typed wrappers (clean and safe)

func CacheWrap1 

func CacheWrap1[T, A any](
	w *CacheWrapper,
	fn func(context.Context, A) (T, error),
	opts ...CacheFuncOption,
) func(context.Context, A) (T, error)

func CacheWrap2 

func CacheWrap2[T, A, B any](
	w *CacheWrapper,
	fn func(context.Context, A, B) (T, error),
	opts ...CacheFuncOption,
) func(context.Context, A, B) (T, error)

func CacheWrap3 

func CacheWrap3[T, A, B, C any](
	w *CacheWrapper,
	fn func(context.Context, A, B, C) (T, error),
	opts ...CacheFuncOption,
) func(context.Context, A, B, C) (T, error)

func CacheWrap4 

func CacheWrap4[T, A, B, C, D any](
	w *CacheWrapper,
	fn func(context.Context, A, B, C, D) (T, error),
	opts ...CacheFuncOption,
) func(context.Context, A, B, C, D) (T, error)

func CacheWrap5 added in v1.0.0 

func CacheWrap5[T, A, B, C, D, E any](
	w *CacheWrapper,
	fn func(context.Context, A, B, C, D, E) (T, error),
	opts ...CacheFuncOption,
) func(context.Context, A, B, C, D, E) (T, error)

func InitDefault 

func InitDefault(opts ...Option)

InitDefault initializes the default package-level client.

func UnmarshalPayloadTyped 

func UnmarshalPayloadTyped[T any](payload []byte) (T, error)

UnmarshalPayloadTyped unmarshals the payload into a specific type

Types

type AtTime added in v1.0.0 

type AtTime struct {
	Hour   uint
	Minute uint
	Second uint
}

func NewAtTime added in v1.0.0 

func NewAtTime(hour, minute, seconds uint) AtTime

func (AtTime) ToGocronAtTime added in v1.0.0 

func (a AtTime) ToGocronAtTime() gocronv2.AtTime

type BackpressureError added in v1.0.0 

type BackpressureError struct {
	Pending, Limit int64
}

BackpressureError 反压错误类型

func (*BackpressureError) Error added in v1.0.0 

func (e *BackpressureError) Error() string

type BackpressureMode added in v1.0.0 

type BackpressureMode int

BackpressureMode 反压控制模式 

const (
	BackpressureBlock  BackpressureMode = iota // 阻塞等待
	BackpressureDelay                          // 延迟生产
	BackpressureReject                         // 拒绝生产
)

type BloomFilter 

type BloomFilter struct {
	// contains filtered or unexported fields
}

BloomFilter represents a probabilistic data structure using Redis.

func NewBloomFilter 

func NewBloomFilter(client *RedisClient, key string) *BloomFilter

NewBloomFilter creates a new bloom filter.

func (*BloomFilter) Add 

func (b *BloomFilter) Add(ctx context.Context, item string) (bool, error)

Add adds an item to the bloom filter. Returns true if the item might have already been in the filter, false otherwise.

func (*BloomFilter) Reset 

func (b *BloomFilter) Reset(ctx context.Context) error

Reset resets the bloom filter.

func (*BloomFilter) SetExpiry 

func (b *BloomFilter) SetExpiry(ctx context.Context, expiry time.Duration) (bool, error)

SetExpiry sets the expiry time for the bloom filter.

type Broadcaster 

type Broadcaster struct {
	// contains filtered or unexported fields
}

func NewBroadcaster 

func NewBroadcaster(opts ...BroadcasterOption) (*Broadcaster, error)

func (*Broadcaster) PublishSimple 

func (b *Broadcaster) PublishSimple(channel string, payload interface{}) error

func (*Broadcaster) RegisterHandler 

func (b *Broadcaster) RegisterHandler(messageType string, handler TaskHandler) error

func (*Broadcaster) RegisterHandlerFunc 

func (b *Broadcaster) RegisterHandlerFunc(messageType string, handlerFunc func(ctx context.Context, payload []byte) error) error

func (*Broadcaster) Shutdown 

func (b *Broadcaster) Shutdown()

func (*Broadcaster) Start 

func (b *Broadcaster) Start(ctx context.Context) error

type BroadcasterConfig 

type BroadcasterConfig struct {
	// contains filtered or unexported fields
}

type BroadcasterOption 

type BroadcasterOption func(*BroadcasterConfig)

func WithChannel 

func WithChannel(channel string) BroadcasterOption

func WithClient 

func WithClient(redisCli *RedisClient) BroadcasterOption

type CacheFuncOption 

type CacheFuncOption func(*CacheFuncOptions)

CacheFuncOption is a functional option for CacheFuncOptions

func WithDisabled 

func WithDisabled(disabled bool) CacheFuncOption

func WithKey 

func WithKey(key string) CacheFuncOption

func WithKeyGenerator 

func WithKeyGenerator(gen func(ctx context.Context, args ...interface{}) string) CacheFuncOption

func WithKeyPrefix 

func WithKeyPrefix(prefix string) CacheFuncOption

func WithRedisClient 

func WithRedisClient(client *RedisClient) CacheFuncOption

func WithTTL 

func WithTTL(ttl time.Duration) CacheFuncOption

type CacheFuncOptions 

type CacheFuncOptions struct {
	TTL           time.Duration
	KeyPrefix     string
	Key           string
	KeyGenerator  func(ctx context.Context, args ...interface{}) string
	InternalCache cache.Cache
	RedisClient   *RedisClient
	Disabled      bool
}

CacheFuncOptions for the wrapped function

func DefaultCacheFuncOptions 

func DefaultCacheFuncOptions() *CacheFuncOptions

func (*CacheFuncOptions) Clone 

func (c *CacheFuncOptions) Clone() *CacheFuncOptions

type CacheStats 

type CacheStats struct {
	Hits          int64
	Misses        int64
	Errors        int64
	CacheAttempts int64
}

CacheStats holds statistics about cache operations

func (CacheStats) String 

func (s CacheStats) String() string

type CacheWrapper 

type CacheWrapper struct {
	// contains filtered or unexported fields
}

CacheWrapper provides a way to wrap functions with caching capabilities

func NewCacheWrapper 

func NewCacheWrapper(opts ...CacheFuncOption) *CacheWrapper

func (*CacheWrapper) Close 

func (w *CacheWrapper) Close()

Close properly cleans up resources used by the CacheWrapper

func (*CacheWrapper) Delete 

func (w *CacheWrapper) Delete(ctx context.Context, key string) error

func (*CacheWrapper) Get 

func (w *CacheWrapper) Get(ctx context.Context, key string, value any) error

func (*CacheWrapper) GetRedisClient added in v1.0.0 

func (w *CacheWrapper) GetRedisClient() *RedisClient

GetRedisClient returns the Redis client if available

func (*CacheWrapper) GetStats 

func (w *CacheWrapper) GetStats() CacheStats

GetStats returns current cache stats

func (*CacheWrapper) ResetCacheByBaseKey 

func (w *CacheWrapper) ResetCacheByBaseKey(ctx context.Context, baseKey string) error

ResetCacheByBaseKey removes all cached entries associated with the given baseKey

func (*CacheWrapper) ResetCacheByKey 

func (w *CacheWrapper) ResetCacheByKey(ctx context.Context, key string) error

ResetCacheByKey removes the cached result for a specific key

func (*CacheWrapper) Set 

func (w *CacheWrapper) Set(ctx context.Context, key string, value any, opts ...CacheFuncOption) error

func (*CacheWrapper) WrapFunc 

func (w *CacheWrapper) WrapFunc(fn interface{}, opts ...CacheFuncOption) interface{}

WrapFunc wraps a function with caching (low-level, interface{})

type Config 

type Config struct {
	// Mode specifies the Redis connection mode (standalone, cluster, or sentinel).
	Mode Mode

	// Common options
	// Password is the password for the Redis server.
	Password string
	// DB is the database to select (not used in cluster mode).
	DB int
	// PoolSize is the maximum number of socket connections.
	PoolSize int
	// MinIdleConns is the minimum number of idle connections.
	MinIdleConns int
	// DialTimeout is the timeout for establishing new connections.
	DialTimeout time.Duration
	// ReadTimeout is the timeout for socket reads.
	ReadTimeout time.Duration
	// WriteTimeout is the timeout for socket writes.
	WriteTimeout time.Duration
	// PoolTimeout is the timeout for getting a connection from the pool.
	PoolTimeout time.Duration
	// IdleTimeout is the timeout for idle connections.
	IdleTimeout time.Duration
	// MaxRetries is the maximum number of retries before giving up.
	MaxRetries int
	// MinRetryBackoff is the minimum backoff between each retry.
	MinRetryBackoff time.Duration
	// MaxRetryBackoff is the maximum backoff between each retry.
	MaxRetryBackoff time.Duration

	// Standalone options
	// Addr is the address of the Redis server (used in standalone mode).
	Addr string

	// Cluster options
	// Addrs is a list of Redis cluster node addresses (used in cluster mode).
	Addrs []string
	// MaxRedirects is the maximum number of redirects to follow (used in cluster mode).
	MaxRedirects int
	// RouteByLatency enables routing read-only commands to the closest master or replica node (used in cluster mode).
	RouteByLatency bool
	// RouteRandomly enables routing read-only commands to random nodes (used in cluster mode).
	RouteRandomly bool

	// Sentinel options
	// MasterName is the name of the master node (used in sentinel mode).
	MasterName string
	// SentinelAddrs is a list of Redis sentinel addresses (used in sentinel mode).
	SentinelAddrs []string
	// SentinelPassword is the password for the sentinel servers (used in sentinel mode).
	SentinelPassword string
}

Config holds the configuration for the Redis client.

func DefaultConfig 

func DefaultConfig() *Config

DefaultConfig returns a default configuration for the Redis client.

type Counter 

type Counter struct {
	// contains filtered or unexported fields
}

Counter represents a distributed counter using Redis.

func NewCounter 

func NewCounter(client *RedisClient, key string) *Counter

NewCounter creates a new distributed counter.

func (*Counter) Decrement 

func (c *Counter) Decrement(ctx context.Context, amount int64) (int64, error)

Decrement decrements the counter by the given amount.

func (*Counter) Get 

func (c *Counter) Get(ctx context.Context) (int64, error)

Get returns the current value of the counter.

func (*Counter) Increment 

func (c *Counter) Increment(ctx context.Context, amount int64) (int64, error)

Increment increments the counter by the given amount.

func (*Counter) Reset 

func (c *Counter) Reset(ctx context.Context) error

Reset resets the counter to zero.

func (*Counter) SetExpiry 

func (c *Counter) SetExpiry(ctx context.Context, expiry time.Duration) (bool, error)

SetExpiry sets the expiry time for the counter.

type CronSchedule added in v1.0.0 

type CronSchedule struct {
	Spec string
}

CronSchedule Cron 调度配置

func (*CronSchedule) GetCronSpec added in v1.0.0 

func (cs *CronSchedule) GetCronSpec() string

GetCronSpec 获取 Cron 表达式（不做转换） 返回原始表达式，由调用者根据 HasSeconds() 决定是否需要处理

func (*CronSchedule) HasSeconds added in v1.0.0 

func (cs *CronSchedule) HasSeconds() bool

HasSeconds 检查 Cron 表达式是否包含秒字段 标准 Cron 格式： - 5 字段：分 时 日 月 周（不支持秒） - 6 字段：秒 分 时 日 月 周（支持秒）

type DLockOption 

type DLockOption func(*DistLockManager)

Option configures the DistLockManager.

func WithCacheSize 

func WithCacheSize(size int) DLockOption

WithCacheSize sets the max number of cached mutexes (default: 1000).

func WithRetryDelay 

func WithRetryDelay(delay time.Duration) DLockOption

WithRetryDelay sets the retry delay for redsync.

func WithTries 

func WithTries(tries int) DLockOption

WithTries sets the max number of attempts to acquire the lock.

type DailySchedule added in v1.0.0 

type DailySchedule struct {
	Interval int
	AtTimes  []AtTime
}

DailySchedule 每日调度配置

type DistLock 

type DistLock struct {
	// contains filtered or unexported fields
}

DistLock is a wrapper that ensures safe unlock and provides metadata.

func (*DistLock) IsLocked 

func (dl *DistLock) IsLocked() bool

IsLocked checks if the lock is still held (best-effort, not 100% reliable).

func (*DistLock) Key 

func (dl *DistLock) Key() string

Key returns the lock key (useful for logging/metrics).

func (*DistLock) Unlock 

func (dl *DistLock) Unlock() (bool, error)

Unlock releases the distributed lock. Returns true if successful, false if lock was lost or already released.

type DistLockManager 

type DistLockManager struct {
	// contains filtered or unexported fields
}

DistLockManager manages distributed locks with caching and config options.

func NewLockManager 

func NewLockManager(client *RedisClient, opts ...DLockOption) (*DistLockManager, error)

NewLockManager creates a new distributed lock manager. It uses an LRU cache to avoid infinite growth of mutex entries.

func (*DistLockManager) GetLock 

func (lm *DistLockManager) GetLock(key string) *redsync.Mutex

GetLock returns a redsync.Mutex for the given key. It caches the mutex instance to avoid recreation. Not goroutine-safe for the same key's mutex usage (caller must manage).

func (*DistLockManager) Lock 

func (lm *DistLockManager) Lock(ctx context.Context, key string, opts ...LockOption) (*DistLock, error)

Lock acquires the lock with a timeout and returns a wrapper for safe unlock. This is the recommended way to use the lock.

type DurationSchedule added in v1.0.0 

type DurationSchedule struct {
	Interval time.Duration
}

DurationSchedule 固定间隔调度配置

type HandlerFunc 

type HandlerFunc func(ctx context.Context, payload []byte) error

HandlerFunc is a function type that implements TaskHandler (duplicated from consume.go for Redis implementation)

func (HandlerFunc) ProcessTask 

func (f HandlerFunc) ProcessTask(ctx context.Context, payload []byte) error

ProcessTask calls the HandlerFunc

type KeyConfig 

type KeyConfig struct {
	Expire     time.Duration `json:"expire"`
	MaxRecords int64         `json:"max_records"`
}

KeyConfig stores configuration for a key.

type KeyNotConfiguredError 

type KeyNotConfiguredError struct {
	Key string
}

KeyNotConfiguredError is returned when a key is not configured.

func (*KeyNotConfiguredError) Error 

func (e *KeyNotConfiguredError) Error() string

type LeaderElector added in v1.0.0 

type LeaderElector struct {
	// contains filtered or unexported fields
}

LeaderElector 支持回调

func NewLeaderElector added in v1.0.0 

func NewLeaderElector(client *RedisClient, instanceID string, opts ...LeaderElectorOption) *LeaderElector

func (*LeaderElector) Campaign added in v1.0.0 

func (e *LeaderElector) Campaign() error

Campaign 尝试竞选

func (*LeaderElector) GetLeaderID added in v1.0.0 

func (e *LeaderElector) GetLeaderID() (string, error)

GetLeaderID 获取当前 leader 的 instanceID

func (*LeaderElector) GetRole added in v1.0.0 

func (e *LeaderElector) GetRole() Role

GetRole 获取当前角色

func (*LeaderElector) IsLeader added in v1.0.0 

func (e *LeaderElector) IsLeader() bool

func (*LeaderElector) RegisterOnRoleChanged added in v1.0.0 

func (e *LeaderElector) RegisterOnRoleChanged(fn OnRoleChanged)

RegisterOnRoleChanged 注册角色变化回调

func (*LeaderElector) RenewLeader added in v1.0.0 

func (e *LeaderElector) RenewLeader() error

RenewLeader 续租（使用 Lua 脚本确保原子性）

func (*LeaderElector) Run added in v1.0.0 

func (e *LeaderElector) Run()

Run 启动选举循环

func (*LeaderElector) Stop added in v1.0.0 

func (e *LeaderElector) Stop()

type LeaderElectorOption added in v1.0.0 

type LeaderElectorOption func(*LeaderElector)

LeaderElectorOption 配置选项

func WithHealthInterval added in v1.0.0 

func WithHealthInterval(interval time.Duration) LeaderElectorOption

WithHealthInterval 设置健康检查间隔

func WithLeaderKey added in v1.0.0 

func WithLeaderKey(key string) LeaderElectorOption

WithLeaderKey 设置 leader key

func WithLeaderTTL added in v1.0.0 

func WithLeaderTTL(ttl time.Duration) LeaderElectorOption

WithLeaderTTL 设置 leader TTL

func WithReelectJitter added in v1.0.0 

func WithReelectJitter(jitter time.Duration) LeaderElectorOption

WithReelectJitter 设置重新选举抖动时间

type LocalConsumer 

type LocalConsumer struct {
	// contains filtered or unexported fields
}

LocalConsumer represents a local pub/sub consumer

func NewLocalConsumer 

func NewLocalConsumer(publisher *LocalPublisher, channel ...string) *LocalConsumer

NewLocalConsumer creates a new local consumer

func (*LocalConsumer) RegisterHandler 

func (c *LocalConsumer) RegisterHandler(messageType string, handler TaskHandler) error

RegisterHandler registers a handler for a specific message type

func (*LocalConsumer) RegisterHandlerFunc 

func (c *LocalConsumer) RegisterHandlerFunc(messageType string, handlerFunc func(ctx context.Context, payload []byte) error) error

RegisterHandlerFunc registers a handler function for a specific message type

func (*LocalConsumer) Shutdown 

func (c *LocalConsumer) Shutdown()

Shutdown gracefully shuts down the consumer

func (*LocalConsumer) Start 

func (c *LocalConsumer) Start(ctx context.Context) error

Start starts the consumer to listen for messages

type LocalMessage 

type LocalMessage struct {
	Type    string          `json:"type"`
	Payload json.RawMessage `json:"payload"`
}

LocalMessage represents a message in the local pub/sub system

type LocalPublisher 

type LocalPublisher struct {
	// contains filtered or unexported fields
}

LocalPublisher represents a local pub/sub publisher

func NewLocalPublisher 

func NewLocalPublisher(channel ...string) *LocalPublisher

NewLocalPublisher creates a new local publisher

func (*LocalPublisher) Close 

func (p *LocalPublisher) Close() error

Close closes the local publisher

func (*LocalPublisher) PublishSimple 

func (p *LocalPublisher) PublishSimple(channel string, payload interface{}) error

PublishSimple publishes a message to a local channel

func (*LocalPublisher) Subscribe 

func (p *LocalPublisher) Subscribe(ch chan LocalMessage)

Subscribe adds a listener for the specified channel

func (*LocalPublisher) Unsubscribe 

func (p *LocalPublisher) Unsubscribe(ch chan LocalMessage)

Unsubscribe removes a listener for the specified channel

type LockOption 

type LockOption func(*lockConfig)

LockOption allows per-lock configuration.

func WithLockExpiry added in v1.0.0 

func WithLockExpiry(expiry time.Duration) LockOption

func WithLockTimeout 

func WithLockTimeout(timeout time.Duration) LockOption

WithLockTimeout sets the maximum time to wait for acquiring the lock.

type Message added in v1.0.0 

type Message struct {
	ID      string      `json:"id"`
	Payload interface{} `json:"payload"`
	Time    time.Time   `json:"time"`
}

Message 队列消息结构

type Mode 

type Mode int

Mode represents the Redis connection mode. 

const (
	// StandaloneMode represents a standalone Redis server.
	StandaloneMode Mode = iota
	// ClusterMode represents a Redis cluster.
	ClusterMode
	// SentinelMode represents a Redis sentinel setup.
	SentinelMode
)

type MonthlySchedule added in v1.0.0 

type MonthlySchedule struct {
	Interval    int
	DaysOfMonth []int
	AtTimes     []AtTime
}

MonthlySchedule 每月调度配置

type OnRoleChanged added in v1.0.0 

type OnRoleChanged func(oldRole, newRole Role)

OnRoleChanged 回调函数类型：oldRole, newRole

type OnetimeSchedule added in v1.0.0 

type OnetimeSchedule struct {
	ExecuteAt time.Time
}

OnetimeSchedule 一次性调度配置

type Option 

type Option func(*Config)

Option defines a function that configures the Redis client.

func WithAddr 

func WithAddr(addr string) Option

WithAddr sets the Redis server address.

func WithAddrs 

func WithAddrs(addrs []string) Option

WithAddrs sets the Redis cluster node addresses.

func WithDB 

func WithDB(db int) Option

WithDB sets the Redis database to select.

func WithDialTimeout 

func WithDialTimeout(dialTimeout time.Duration) Option

WithDialTimeout sets the timeout for establishing new connections.

func WithIdleTimeout 

func WithIdleTimeout(idleTimeout time.Duration) Option

WithIdleTimeout sets the timeout for idle connections.

func WithMasterName 

func WithMasterName(masterName string) Option

WithMasterName sets the name of the master node for sentinel mode.

func WithMaxRedirects 

func WithMaxRedirects(maxRedirects int) Option

WithMaxRedirects sets the maximum number of redirects to follow.

func WithMaxRetries 

func WithMaxRetries(maxRetries int) Option

WithMaxRetries sets the maximum number of retries before giving up.

func WithMaxRetryBackoff 

func WithMaxRetryBackoff(maxRetryBackoff time.Duration) Option

WithMaxRetryBackoff sets the maximum backoff between each retry.

func WithMinIdleConns 

func WithMinIdleConns(minIdleConns int) Option

WithMinIdleConns sets the minimum number of idle connections.

func WithMinRetryBackoff 

func WithMinRetryBackoff(minRetryBackoff time.Duration) Option

WithMinRetryBackoff sets the minimum backoff between each retry.

func WithMode 

func WithMode(mode Mode) Option

WithMode sets the Redis connection mode.

func WithPassword 

func WithPassword(password string) Option

WithPassword sets the Redis server password.

func WithPoolSize 

func WithPoolSize(poolSize int) Option

WithPoolSize sets the maximum number of socket connections.

func WithPoolTimeout 

func WithPoolTimeout(poolTimeout time.Duration) Option

WithPoolTimeout sets the timeout for getting a connection from the pool.

func WithReadTimeout 

func WithReadTimeout(readTimeout time.Duration) Option

WithReadTimeout sets the timeout for socket reads.

func WithRouteByLatency 

func WithRouteByLatency(routeByLatency bool) Option

WithRouteByLatency enables routing read-only commands to the closest master or replica node.

func WithRouteRandomly 

func WithRouteRandomly(routeRandomly bool) Option

WithRouteRandomly enables routing read-only commands to random nodes.

func WithSentinelAddrs 

func WithSentinelAddrs(sentinelAddrs []string) Option

WithSentinelAddrs sets the Redis sentinel addresses.

func WithSentinelPassword 

func WithSentinelPassword(sentinelPassword string) Option

WithSentinelPassword sets the password for the sentinel servers.

func WithWriteTimeout 

func WithWriteTimeout(writeTimeout time.Duration) Option

WithWriteTimeout sets the timeout for socket writes.

type PubSub 

type PubSub struct {
	// contains filtered or unexported fields
}

PubSub represents a publish/subscribe system using Redis.

func NewPubSub 

func NewPubSub(client *RedisClient) *PubSub

NewPubSub creates a new publish/subscribe system.

func (*PubSub) Publish 

func (p *PubSub) Publish(ctx context.Context, channel string, message interface{}) error

Publish publishes a message to the given channel.

func (*PubSub) Subscribe 

func (p *PubSub) Subscribe(ctx context.Context, channels ...string) *redis.PubSub

Subscribe subscribes to the given channels and returns a subscription.

type QueueMetrics added in v1.0.0 

type QueueMetrics struct {
	ProducedTotal    atomic.Int64
	ConsumedTotal    atomic.Int64
	ProcessedTotal   atomic.Int64
	FailedTotal      atomic.Int64
	RetriedTotal     atomic.Int64
	BackpressureHits atomic.Int64
}

QueueMetrics 队列指标

type RandomSchedule added in v1.0.0 

type RandomSchedule struct {
	IntervalMin time.Duration
	IntervalMax time.Duration
}

RandomSchedule 随机间隔调度配置

type RateLimiter 

type RateLimiter struct {
	// contains filtered or unexported fields
}

RateLimiter represents a rate limiter using Redis.

func NewRateLimiter 

func NewRateLimiter(client *RedisClient, key string, limit int, window time.Duration) *RateLimiter

NewRateLimiter creates a new rate limiter.

func (*RateLimiter) Allow 

func (r *RateLimiter) Allow(ctx context.Context) (bool, error)

Allow checks if the action is allowed by the rate limiter. Returns true if the action is allowed, false otherwise.

func (*RateLimiter) Reset 

func (r *RateLimiter) Reset(ctx context.Context) error

Reset resets the rate limiter.

type Record 

type Record[T any] struct {
	Data      T         `json:"data"`
	Timestamp time.Time `json:"timestamp"`
}

Record holds the data and timestamp.

type RecordStore 

type RecordStore[T any] struct {
	// contains filtered or unexported fields
}

RecordStore is a generic in-memory store for time-series records.

func NewRecordStore 

func NewRecordStore[T any](client *RedisClient, namespace string) *RecordStore[T]

NewR'e'a'co'r'dStore creates a new RecordStore instance.

func (*RecordStore[T]) CreateKey 

func (ms *RecordStore[T]) CreateKey(ctx context.Context, key string, expire time.Duration, maxRecords int64) error

CreateKey creates a new key with config (idempotent). Uses SETNX to prevent race conditions in multi-instance environments.

func (*RecordStore[T]) DeleteKey 

func (ms *RecordStore[T]) DeleteKey(ctx context.Context, key string) (int64, int64, error)

DeleteKey removes the key and its config.

func (*RecordStore[T]) GetLatestRecords 

func (ms *RecordStore[T]) GetLatestRecords(ctx context.Context, key string, count int64) ([]Record[T], error)

GetLatestRecords retrieves the latest N records.

func (*RecordStore[T]) GetRecordCount 

func (ms *RecordStore[T]) GetRecordCount(ctx context.Context, key string) (int64, error)

GetRecordCount returns the number of records for a key.

func (*RecordStore[T]) IsKeyCreated 

func (ms *RecordStore[T]) IsKeyCreated(ctx context.Context, key string) (bool, error)

IsKeyCreated checks if the key has been created (i.e., config exists). Uses local cache for performance.

func (*RecordStore[T]) SetConfigCacheTTL 

func (ms *RecordStore[T]) SetConfigCacheTTL(ttl time.Duration)

SetConfigCacheTTL sets the TTL for config cache.

func (*RecordStore[T]) WriteRecord 

func (ms *RecordStore[T]) WriteRecord(ctx context.Context, key string, data T) error

WriteRecord writes a new record to the key.

type RedisClient 

type RedisClient struct {
	// contains filtered or unexported fields
}

RedisClient is a wrapper around the Redis client.

func Default 

func Default() *RedisClient

Default returns the default client.

func New 

func New(opts ...Option) *RedisClient

New creates a new RedisClient instance with the given options.

func (*RedisClient) Client 

func (c *RedisClient) Client() *redis.Client

Client returns the underlying Redis client.

func (*RedisClient) Close 

func (c *RedisClient) Close() error

Close closes the client, releasing any open resources.

func (*RedisClient) Decr 

func (c *RedisClient) Decr(ctx context.Context, key string) (int64, error)

Decr decrements the integer value of a key by one.

func (*RedisClient) DecrBy 

func (c *RedisClient) DecrBy(ctx context.Context, key string, value int64) (int64, error)

DecrBy decrements the integer value of a key by the given amount.

func (*RedisClient) Del 

func (c *RedisClient) Del(ctx context.Context, key string) (int64, error)

Del deletes one or more keys.

func (*RedisClient) Eval 

func (c *RedisClient) Eval(ctx context.Context, script string, keys []string, args ...interface{}) (interface{}, error)

Eval evaluates a Lua script.

func (*RedisClient) Exists 

func (c *RedisClient) Exists(ctx context.Context, keys ...string) (int64, error)

Exists checks if one or more keys exist.

func (*RedisClient) Expire 

func (c *RedisClient) Expire(ctx context.Context, key string, expiration time.Duration) (bool, error)

Expire sets the expiration for a key.

func (*RedisClient) Get 

func (c *RedisClient) Get(ctx context.Context, key string) (string, error)

Get gets the value of a key.

func (*RedisClient) GetBytes 

func (c *RedisClient) GetBytes(ctx context.Context, key string) ([]byte, error)

GetBytes gets the value of a key as bytes.

func (*RedisClient) GetObject 

func (c *RedisClient) GetObject(ctx context.Context, key string, value interface{}) error

GetObject gets the value of a key and unmarshals it into the given object.

func (*RedisClient) HDel 

func (c *RedisClient) HDel(ctx context.Context, key string, fields ...string) (int64, error)

HDel deletes one or more hash fields.

func (*RedisClient) HExists 

func (c *RedisClient) HExists(ctx context.Context, key, field string) (bool, error)

HExists returns if field is an existing field in the hash stored at key.

func (*RedisClient) HGet 

func (c *RedisClient) HGet(ctx context.Context, key, field string) (string, error)

HGet returns the value associated with field in the hash stored at key.

func (*RedisClient) HGetAll 

func (c *RedisClient) HGetAll(ctx context.Context, key string) (map[string]string, error)

HGetAll returns all fields and values of the hash stored at key.

func (*RedisClient) HKeys 

func (c *RedisClient) HKeys(ctx context.Context, key string) ([]string, error)

HKeys returns all field names in the hash stored at key.

func (*RedisClient) HLen 

func (c *RedisClient) HLen(ctx context.Context, key string) (int64, error)

HLen returns the number of fields in the hash stored at key.

func (*RedisClient) HSet 

func (c *RedisClient) HSet(ctx context.Context, key, field string, value interface{}) error

HSet sets field in the hash stored at key to value.

func (*RedisClient) Incr 

func (c *RedisClient) Incr(ctx context.Context, key string) (int64, error)

Incr increments the integer value of a key by one.

func (*RedisClient) IncrBy 

func (c *RedisClient) IncrBy(ctx context.Context, key string, value int64) (int64, error)

IncrBy increments the integer value of a key by the given amount.

func (*RedisClient) LLen 

func (c *RedisClient) LLen(ctx context.Context, key string) (int64, error)

LLen returns the length of the list stored at key.

func (*RedisClient) LPop 

func (c *RedisClient) LPop(ctx context.Context, key string) (string, error)

LPop removes and returns the first element of the list stored at key.

func (*RedisClient) LPush 

func (c *RedisClient) LPush(ctx context.Context, key string, values ...interface{}) (int64, error)

LPush inserts all the specified values at the head of the list stored at key.

func (*RedisClient) LRange 

func (c *RedisClient) LRange(ctx context.Context, key string, start, stop int64) ([]string, error)

LRange returns the specified elements of the list stored at key.

func (*RedisClient) MGet 

func (r *RedisClient) MGet(ctx context.Context, keys ...string) (map[string]any, error)

func (*RedisClient) MSet 

func (r *RedisClient) MSet(ctx context.Context, value map[string]any, expire time.Duration) error

func (*RedisClient) Nil 

func (r *RedisClient) Nil() error

func (*RedisClient) Ping 

func (c *RedisClient) Ping(ctx context.Context) (string, error)

Ping pings the Redis server.

func (*RedisClient) Pipeline 

func (c *RedisClient) Pipeline() redis.Pipeliner

Pipeline creates a new pipeline.

func (*RedisClient) Publish 

func (c *RedisClient) Publish(ctx context.Context, channel string, message interface{}) (int64, error)

Publish publishes a message to the specified channel.

func (*RedisClient) RPop 

func (c *RedisClient) RPop(ctx context.Context, key string) (string, error)

RPop removes and returns the last element of the list stored at key.

func (*RedisClient) RPush 

func (c *RedisClient) RPush(ctx context.Context, key string, values ...interface{}) (int64, error)

RPush inserts all the specified values at the tail of the list stored at key.

func (*RedisClient) SAdd 

func (c *RedisClient) SAdd(ctx context.Context, key string, members ...interface{}) (int64, error)

SAdd adds one or more members to a set.

func (*RedisClient) SCard 

func (c *RedisClient) SCard(ctx context.Context, key string) (int64, error)

SCard returns the set cardinality (number of elements) of the set stored at key.

func (*RedisClient) SIsMember 

func (c *RedisClient) SIsMember(ctx context.Context, key string, member interface{}) (bool, error)

SIsMember returns if member is a member of the set stored at key.

func (*RedisClient) SMembers 

func (c *RedisClient) SMembers(ctx context.Context, key string) ([]string, error)

SMembers returns all the members of the set value stored at key.

func (*RedisClient) SRem 

func (c *RedisClient) SRem(ctx context.Context, key string, members ...interface{}) (int64, error)

SRem removes one or more members from a set.

func (*RedisClient) ScriptLoad 

func (c *RedisClient) ScriptLoad(ctx context.Context, script string) (string, error)

ScriptLoad loads a Lua script into the scripts cache.

func (*RedisClient) Set 

func (c *RedisClient) Set(ctx context.Context, key string, value interface{}, expiration time.Duration) error

Set sets the value of a key.

func (*RedisClient) SetEX 

func (r *RedisClient) SetEX(ctx context.Context, key string, value any, expire time.Duration) error

func (*RedisClient) SetNX 

func (r *RedisClient) SetNX(ctx context.Context, key string, value any, expire time.Duration) (val bool, err error)

func (*RedisClient) SetObject 

func (c *RedisClient) SetObject(ctx context.Context, key string, value interface{}, expiration time.Duration) error

SetObject sets the value of a key to the marshaled object.

func (*RedisClient) SetXX 

func (r *RedisClient) SetXX(ctx context.Context, key string, value any, expire time.Duration) (val bool, err error)

func (*RedisClient) Subscribe 

func (c *RedisClient) Subscribe(ctx context.Context, channels ...string) *redis.PubSub

Subscribe subscribes to the specified channels.

func (*RedisClient) TTL 

func (c *RedisClient) TTL(ctx context.Context, key string) (time.Duration, error)

TTL returns the remaining time to live of a key.

func (*RedisClient) TxPipeline 

func (c *RedisClient) TxPipeline() redis.Pipeliner

TxPipeline creates a new transaction pipeline.

func (*RedisClient) UniversalClient 

func (c *RedisClient) UniversalClient() redis.UniversalClient

UniversalClient returns the underlying universal Redis client.

func (*RedisClient) Watch 

func (c *RedisClient) Watch(ctx context.Context, fn func(*redis.Tx) error, keys ...string) error

Watch watches the given keys to determine execution of the MULTI/EXEC block.

func (*RedisClient) XAck added in v1.0.0 

func (c *RedisClient) XAck(ctx context.Context, stream, group string, ids ...string) (int64, error)

XAck acknowledges one or more messages.

func (*RedisClient) XAdd added in v1.0.0 

func (c *RedisClient) XAdd(ctx context.Context, args *redis.XAddArgs) (string, error)

XAdd adds a new entry to a stream.

func (*RedisClient) XAutoClaim added in v1.0.0 

func (c *RedisClient) XAutoClaim(ctx context.Context, args *redis.XAutoClaimArgs) ([]redis.XMessage, string, error)

XAutoClaim automatically claims pending messages.

func (*RedisClient) XAutoClaimJustID added in v1.0.0 

func (c *RedisClient) XAutoClaimJustID(ctx context.Context, args *redis.XAutoClaimArgs) ([]string, string, error)

XAutoClaimJustID automatically claims pending messages and returns only IDs.

func (*RedisClient) XClaim added in v1.0.0 

func (c *RedisClient) XClaim(ctx context.Context, args *redis.XClaimArgs) ([]redis.XMessage, error)

XClaim claims pending messages.

func (*RedisClient) XClaimJustID added in v1.0.0 

func (c *RedisClient) XClaimJustID(ctx context.Context, args *redis.XClaimArgs) ([]string, error)

XClaimJustID claims pending messages and returns only IDs.

func (*RedisClient) XDel added in v1.0.0 

func (c *RedisClient) XDel(ctx context.Context, stream string, ids ...string) (int64, error)

XDel deletes entries from a stream.

func (*RedisClient) XGroupCreate added in v1.0.0 

func (c *RedisClient) XGroupCreate(ctx context.Context, stream, group, start string) error

XGroupCreate creates a new consumer group.

func (*RedisClient) XGroupCreateConsumer added in v1.0.0 

func (c *RedisClient) XGroupCreateConsumer(ctx context.Context, stream, group, consumer string) (int64, error)

XGroupCreateConsumer creates a consumer in a consumer group.

func (*RedisClient) XGroupCreateMkStream added in v1.0.0 

func (c *RedisClient) XGroupCreateMkStream(ctx context.Context, stream, group, start string) error

XGroupCreateMkStream creates a new consumer group and the stream if it doesn't exist.

func (*RedisClient) XGroupDelConsumer added in v1.0.0 

func (c *RedisClient) XGroupDelConsumer(ctx context.Context, stream, group, consumer string) (int64, error)

XGroupDelConsumer deletes a consumer from a consumer group.

func (*RedisClient) XGroupDestroy added in v1.0.0 

func (c *RedisClient) XGroupDestroy(ctx context.Context, stream, group string) (int64, error)

XGroupDestroy destroys a consumer group.

func (*RedisClient) XGroupSetID added in v1.0.0 

func (c *RedisClient) XGroupSetID(ctx context.Context, stream, group, start string) error

XGroupSetID sets the consumer group last delivered ID.

func (*RedisClient) XInfoConsumers added in v1.0.0 

func (c *RedisClient) XInfoConsumers(ctx context.Context, stream, group string) ([]redis.XInfoConsumer, error)

XInfoConsumers returns information about consumers in a consumer group.

func (*RedisClient) XInfoGroups added in v1.0.0 

func (c *RedisClient) XInfoGroups(ctx context.Context, stream string) ([]redis.XInfoGroup, error)

XInfoGroups returns information about consumer groups.

func (*RedisClient) XInfoStream added in v1.0.0 

func (c *RedisClient) XInfoStream(ctx context.Context, stream string) (*redis.XInfoStream, error)

XInfoStream returns information about a stream.

func (*RedisClient) XInfoStreamFull added in v1.0.0 

func (c *RedisClient) XInfoStreamFull(ctx context.Context, stream string, count int) (*redis.XInfoStreamFull, error)

XInfoStreamFull returns full information about a stream.

func (*RedisClient) XLen added in v1.0.0 

func (c *RedisClient) XLen(ctx context.Context, stream string) (int64, error)

XLen returns the number of entries in a stream.

func (*RedisClient) XPending added in v1.0.0 

func (c *RedisClient) XPending(ctx context.Context, stream, group string) (*redis.XPending, error)

XPending returns pending messages information.

func (*RedisClient) XPendingExt added in v1.0.0 

func (c *RedisClient) XPendingExt(ctx context.Context, args *redis.XPendingExtArgs) ([]redis.XPendingExt, error)

XPendingExt returns extended pending messages information.

func (*RedisClient) XRange added in v1.0.0 

func (c *RedisClient) XRange(ctx context.Context, stream, start, stop string) ([]redis.XMessage, error)

XRange returns a range of entries from a stream.

func (*RedisClient) XRangeN added in v1.0.0 

func (c *RedisClient) XRangeN(ctx context.Context, stream, start, stop string, count int64) ([]redis.XMessage, error)

XRangeN returns a range of entries from a stream with a count limit.

func (*RedisClient) XRead added in v1.0.0 

func (c *RedisClient) XRead(ctx context.Context, args *redis.XReadArgs) ([]redis.XStream, error)

XRead reads data from one or more streams.

func (*RedisClient) XReadGroup added in v1.0.0 

func (c *RedisClient) XReadGroup(ctx context.Context, args *redis.XReadGroupArgs) ([]redis.XStream, error)

XReadGroup reads data from a stream using a consumer group.

func (*RedisClient) XReadStreams added in v1.0.0 

func (c *RedisClient) XReadStreams(ctx context.Context, streams ...string) ([]redis.XStream, error)

XReadStreams reads data from one or more streams using a simple interface.

func (*RedisClient) XRevRange added in v1.0.0 

func (c *RedisClient) XRevRange(ctx context.Context, stream, start, stop string) ([]redis.XMessage, error)

XRevRange returns a range of entries from a stream in reverse order.

func (*RedisClient) XRevRangeN added in v1.0.0 

func (c *RedisClient) XRevRangeN(ctx context.Context, stream, start, stop string, count int64) ([]redis.XMessage, error)

XRevRangeN returns a range of entries from a stream in reverse order with a count limit.

func (*RedisClient) XTrimMaxLen added in v1.0.0 

func (c *RedisClient) XTrimMaxLen(ctx context.Context, stream string, maxLen int64) (int64, error)

XTrimMaxLen trims a stream to a maximum length.

func (*RedisClient) XTrimMaxLenApprox added in v1.0.0 

func (c *RedisClient) XTrimMaxLenApprox(ctx context.Context, stream string, maxLen, limit int64) (int64, error)

XTrimMaxLenApprox trims a stream to approximately a maximum length.

func (*RedisClient) XTrimMinID added in v1.0.0 

func (c *RedisClient) XTrimMinID(ctx context.Context, stream string, minID string) (int64, error)

XTrimMinID trims a stream by minimum ID.

func (*RedisClient) XTrimMinIDApprox added in v1.0.0 

func (c *RedisClient) XTrimMinIDApprox(ctx context.Context, stream string, minID string, limit int64) (int64, error)

XTrimMinIDApprox trims a stream by minimum ID approximately.

func (*RedisClient) ZAdd 

func (c *RedisClient) ZAdd(ctx context.Context, key string, members ...redis.Z) (int64, error)

ZAdd adds one or more members to a sorted set, or updates its score if it already exists.

func (*RedisClient) ZCard 

func (c *RedisClient) ZCard(ctx context.Context, key string) (int64, error)

ZCard returns the sorted set cardinality (number of elements) of the sorted set stored at key.

func (*RedisClient) ZCount added in v1.0.0 

func (c *RedisClient) ZCount(ctx context.Context, key, min, max string) (int64, error)

ZCount returns the number of elements in the sorted set at key with a score between min and max.

func (*RedisClient) ZIncrBy added in v1.0.0 

func (c *RedisClient) ZIncrBy(ctx context.Context, key string, increment float64, member string) (float64, error)

ZIncrBy increments the score of member in the sorted set stored at key by increment.

func (*RedisClient) ZRange 

func (c *RedisClient) ZRange(ctx context.Context, key string, start, stop int64) ([]string, error)

ZRange returns a range of members in a sorted set, by index.

func (*RedisClient) ZRangeByScore added in v1.0.0 

func (c *RedisClient) ZRangeByScore(ctx context.Context, key string, opt *redis.ZRangeBy) ([]string, error)

ZRangeByScore returns all the elements in the sorted set at key with a score between min and max (inclusive).

func (*RedisClient) ZRangeByScoreWithScores added in v1.0.0 

func (c *RedisClient) ZRangeByScoreWithScores(ctx context.Context, key string, opt *redis.ZRangeBy) ([]redis.Z, error)

ZRangeByScoreWithScores returns all the elements with scores in the sorted set at key with a score between min and max (inclusive).

func (*RedisClient) ZRangeWithScores 

func (c *RedisClient) ZRangeWithScores(ctx context.Context, key string, start, stop int64) ([]redis.Z, error)

ZRangeWithScores returns a range of members with scores in a sorted set, by index.

func (*RedisClient) ZRank 

func (c *RedisClient) ZRank(ctx context.Context, key, member string) (int64, error)

ZRank returns the rank of member in the sorted set stored at key, with the scores ordered from low to high.

func (*RedisClient) ZRem 

func (c *RedisClient) ZRem(ctx context.Context, key string, members ...interface{}) (int64, error)

ZRem removes one or more members from a sorted set.

func (*RedisClient) ZRemRangeByRank added in v1.0.0 

func (c *RedisClient) ZRemRangeByRank(ctx context.Context, key string, start, stop int64) (int64, error)

ZRemRangeByRank removes all elements in the sorted set stored at key with rank between start and stop.

func (*RedisClient) ZRemRangeByScore added in v1.0.0 

func (c *RedisClient) ZRemRangeByScore(ctx context.Context, key, min, max string) (int64, error)

ZRemRangeByScore removes all elements in the sorted set stored at key with a score between min and max (inclusive).

func (*RedisClient) ZRevRange added in v1.0.0 

func (c *RedisClient) ZRevRange(ctx context.Context, key string, start, stop int64) ([]string, error)

ZRevRange returns a range of members in a sorted set, by index, with scores ordered from high to low.

func (*RedisClient) ZRevRangeWithScores added in v1.0.0 

func (c *RedisClient) ZRevRangeWithScores(ctx context.Context, key string, start, stop int64) ([]redis.Z, error)

ZRevRangeWithScores returns a range of members with scores in a sorted set, by index, with scores ordered from high to low.

func (*RedisClient) ZRevRank added in v1.0.0 

func (c *RedisClient) ZRevRank(ctx context.Context, key, member string) (int64, error)

ZRevRank returns the rank of member in the sorted set stored at key, with the scores ordered from high to low.

func (*RedisClient) ZScore 

func (c *RedisClient) ZScore(ctx context.Context, key, member string) (float64, error)

ZScore returns the score of member in the sorted set at key.

type RedisConsumer 

type RedisConsumer struct {
	// contains filtered or unexported fields
}

RedisConsumer represents a Redis pub/sub consumer

func NewRedisConsumer 

func NewRedisConsumer(redisCli *RedisClient, channel ...string) (*RedisConsumer, error)

NewRedisConsumerWithConfig creates a new Redis consumer using application configuration

func (*RedisConsumer) RegisterHandler 

func (c *RedisConsumer) RegisterHandler(messageType string, handler TaskHandler) error

RegisterHandler registers a handler for a specific message type

func (*RedisConsumer) RegisterHandlerFunc 

func (c *RedisConsumer) RegisterHandlerFunc(messageType string, handlerFunc func(ctx context.Context, payload []byte) error) error

RegisterHandlerFunc registers a handler function for a specific message type

func (*RedisConsumer) Shutdown 

func (c *RedisConsumer) Shutdown()

Shutdown gracefully shuts down the consumer

func (*RedisConsumer) Start 

func (c *RedisConsumer) Start(ctx context.Context) error

Start starts the consumer to listen for messages

type RedisPublisher 

type RedisPublisher struct {
	// contains filtered or unexported fields
}

RedisPublisher represents a Redis pub/sub publisher

func NewRedisPublisher 

func NewRedisPublisher(redisCli *RedisClient, channel ...string) (*RedisPublisher, error)

NewRedisPublisher creates a new Redis publisher using application configuration

func (*RedisPublisher) Close 

func (p *RedisPublisher) Close() error

Close closes the Redis client connection

func (*RedisPublisher) PublishSimple 

func (p *RedisPublisher) PublishSimple(channel string, payload interface{}) error

PublishSimple publishes a message to a Redis channel with default options

type Role added in v1.0.0 

type Role string
const (
	RoleLeader   Role = "leader"
	RoleFollower Role = "follower"
)

type Scheduler added in v1.0.0 

type Scheduler struct {
	// contains filtered or unexported fields
}

Scheduler 分布式调度器

func NewScheduler added in v1.0.0 

func NewScheduler(redis *RedisClient, opts ...SchedulerOption) *Scheduler

NewScheduler 创建分布式调度器

func (*Scheduler) AddTask added in v1.0.0 

func (s *Scheduler) AddTask(definition *TaskDefinition, opts ...TaskOption) error

AddTask 添加一个分布式任务

func (*Scheduler) GetInstanceID added in v1.0.0 

func (s *Scheduler) GetInstanceID() string

func (*Scheduler) GetNamespace added in v1.0.0 

func (s *Scheduler) GetNamespace() string

func (*Scheduler) GetRole added in v1.0.0 

func (s *Scheduler) GetRole() Role

func (*Scheduler) GetTaskStatus added in v1.0.0 

func (s *Scheduler) GetTaskStatus(taskID string) (*TaskStatus, error)

GetTaskStatus 获取任务状态 在分布式环境下，所有节点返回一致的状态信息：

  • 状态信息从 taskStatus map 中读取（通过 TaskEventStatus 事件同步）
  • 如果状态不存在，则创建新的状态对象
  • 所有节点都能获取完整的状态信息，确保一致性

func (*Scheduler) IsLeader added in v1.0.0 

func (s *Scheduler) IsLeader() bool

func (*Scheduler) ListTasks added in v1.0.0 

func (s *Scheduler) ListTasks() []string

func (*Scheduler) ListTasksByTag added in v1.0.0 

func (s *Scheduler) ListTasksByTag(tag string) []string

ListTasksByTag 按标签查询任务

func (*Scheduler) RegisterFactory added in v1.0.0 

func (s *Scheduler) RegisterFactory(key string, factory TaskHandlerFactory)

RegisterFactory 注册一个工厂，使用 key 作为标识 在 AddTask 或 UpdateTask 时通过 WithTaskFactoryKey(key) 来指定使用哪个工厂

func (*Scheduler) RemoveTask added in v1.0.0 

func (s *Scheduler) RemoveTask(taskID string, opts ...TaskOption) bool

func (*Scheduler) Start added in v1.0.0 

func (s *Scheduler) Start(opts ...TaskOption)

func (*Scheduler) Stop added in v1.0.0 

func (s *Scheduler) Stop(opts ...TaskOption)

func (*Scheduler) StopRunningTask added in v1.0.0 

func (s *Scheduler) StopRunningTask(taskID string) bool

func (*Scheduler) UnregisterFactory added in v1.0.0 

func (s *Scheduler) UnregisterFactory(key string)

UnregisterFactory 注销指定 key 的工厂

func (*Scheduler) UpdateTask added in v1.0.0 

func (s *Scheduler) UpdateTask(definition *TaskDefinition, opts ...TaskOption) error

UpdateTask 更新一个现有的分布式任务

type SchedulerOption added in v1.0.0 

type SchedulerOption func(*Scheduler)

SchedulerOption 配置选项

func WithSchedulerInstanceID added in v1.0.0 

func WithSchedulerInstanceID(instanceID string) SchedulerOption

func WithSchedulerLockKeyPrefix added in v1.0.0 

func WithSchedulerLockKeyPrefix(prefix string) SchedulerOption

func WithSchedulerLockWaitTimeout added in v1.0.0 

func WithSchedulerLockWaitTimeout(timeout time.Duration) SchedulerOption

func WithSchedulerMaxConcurrency added in v1.0.0 

func WithSchedulerMaxConcurrency(n int) SchedulerOption

func WithSchedulerNamespace added in v1.0.0 

func WithSchedulerNamespace(namespace string) SchedulerOption

func WithSchedulerQueueConsumeConcurrency added in v1.0.0 

func WithSchedulerQueueConsumeConcurrency(n int) SchedulerOption

type StreamQueue added in v1.0.0 

type StreamQueue struct {
	// contains filtered or unexported fields
}

StreamQueue 基于 Redis Streams 的队列（带反压控制）

func NewStreamQueue added in v1.0.0 

func NewStreamQueue(client *RedisClient, stream, group, consumer string) *StreamQueue

NewStreamQueue 创建队列（默认不启用反压）

func (*StreamQueue) Close added in v1.0.0 

func (q *StreamQueue) Close() error

Close 优雅关闭

func (*StreamQueue) Consume added in v1.0.0 

func (q *StreamQueue) Consume(handler func(*Message) error)

Consume 启动消费者

func (*StreamQueue) EnableBackpressure added in v1.0.0 

func (q *StreamQueue) EnableBackpressure(mode BackpressureMode, maxPending int, checkInterval time.Duration) *StreamQueue

EnableBackpressure 启用反压控制

func (*StreamQueue) GetMetrics added in v1.0.0 

func (q *StreamQueue) GetMetrics() *QueueMetrics

GetMetrics 获取队列指标

func (*StreamQueue) Produce added in v1.0.0 

func (q *StreamQueue) Produce(ctx context.Context, payload interface{}) error

Produce 生产消息（带反压控制）

func (*StreamQueue) SetBatchSize added in v1.0.0 

func (q *StreamQueue) SetBatchSize(size int) *StreamQueue

SetBatchSize 设置批处理大小

func (*StreamQueue) SetConsumeConcurrency added in v1.0.0 

func (q *StreamQueue) SetConsumeConcurrency(n int) *StreamQueue

func (*StreamQueue) SetRetryConfig added in v1.0.0 

func (q *StreamQueue) SetRetryConfig(claimInterval, claimMinIdle time.Duration, maxRetries int) *StreamQueue

SetRetryConfig 设置重试配置

type Task 

type Task struct {
	TaskDefinition
	Job     gocronv2.Job
	Handler func(context.Context) error
}

Task 是可调度的任务，包含核心字段和处理器

type TaskCore added in v1.0.0 

type TaskCore struct {
	ID                string
	Spec              string // Cron 表达式（当 ScheduleType="cron" 时使用）
	FactoryKey        string // 工厂的注册 key（必需）
	Timeout           time.Duration
	StartTime         *time.Time
	Deadline          *time.Time
	StartImmediately  bool
	ScheduleType      string        // "duration", "random", "cron", "daily", "weekly", "monthly", "onetime"
	Interval          time.Duration // 固定间隔（ScheduleType="duration" 时使用）
	IntervalMin       time.Duration // 随机最小值（ScheduleType="random" 时使用）
	IntervalMax       time.Duration // 随机最大值（ScheduleType="random" 时使用）
	DayInterval       int           // 天/周/月 间隔（ScheduleType="daily/weekly/monthly" 时使用）
	DaysOfWeek        []int         // 周几（0=周日，ScheduleType="weekly" 时使用）
	DaysOfMonth       []int         // 月几（ScheduleType="monthly" 时使用）
	AtTimes           []AtTime      // 执行时间（ScheduleType="daily/weekly/monthly" 时使用）
	IntervalMode      string        // "from-scheduled" 或 "from-completion"
	LastDayOnly       bool          // 最后一天执行标志（@monthlyL）
	LastDayOffset     int           // 倒数第 x 天偏移（@monthlyLx/@monthlyL-<n>），0 表示最后一天
	LastBusinessDay   bool          // 最后一个工作日标志（@monthlyLB）
	MonthlyNthWeekday int           // 每月第 N 个某星期几的 weekday（0=周日..6=周六），用于 @monthlyW<d>/<n>/
	MonthlyNthIndex   int           // 每月第 N 次出现的序号（>=1），用于 @monthlyW<d>/<n>/
	Tags              []string      // 任务标签列表
	Description       string        // 任务描述
	MaxConcurrency    int           // 任务最大并发数（0表示不限制）
}

TaskCore 任务核心字段 - 被 TaskDefinition 和 TaskEvent 共享

func (*TaskCore) GetAtTimes added in v1.0.0 

func (tc *TaskCore) GetAtTimes() gocronv2.AtTimes

func (*TaskCore) GetCronSchedule added in v1.0.0 

func (tc *TaskCore) GetCronSchedule() *CronSchedule

GetCronSchedule 获取 Cron 调度配置

func (*TaskCore) GetDailySchedule added in v1.0.0 

func (tc *TaskCore) GetDailySchedule() *DailySchedule

GetDailySchedule 获取每日调度配置

func (*TaskCore) GetDurationSchedule added in v1.0.0 

func (tc *TaskCore) GetDurationSchedule() *DurationSchedule

GetDurationSchedule 获取固定间隔调度配置

func (*TaskCore) GetMonthlySchedule added in v1.0.0 

func (tc *TaskCore) GetMonthlySchedule() *MonthlySchedule

GetMonthlySchedule 获取每月调度配置

func (*TaskCore) GetOnetimeSchedule added in v1.0.0 

func (tc *TaskCore) GetOnetimeSchedule() *OnetimeSchedule

GetOnetimeSchedule 获取一次性调度配置

func (*TaskCore) GetRandomSchedule added in v1.0.0 

func (tc *TaskCore) GetRandomSchedule() *RandomSchedule

GetRandomSchedule 获取随机间隔调度配置

func (*TaskCore) GetWeeklySchedule added in v1.0.0 

func (tc *TaskCore) GetWeeklySchedule() *WeeklySchedule

GetWeeklySchedule 获取每周调度配置

type TaskDefOption added in v1.0.0 

type TaskDefOption func(*TaskDefinition)

func IntervalFromCompletion added in v1.0.0 

func IntervalFromCompletion() TaskDefOption

func StartNow added in v1.0.0 

func StartNow() TaskDefOption

func WithDeadline added in v1.0.0 

func WithDeadline(t time.Time) TaskDefOption

func WithDefDeadline added in v1.0.0 

func WithDefDeadline(t time.Time) TaskDefOption

func WithDefDescription added in v1.0.0 

func WithDefDescription(desc string) TaskDefOption

func WithDefIntervalFromCompletion added in v1.0.0 

func WithDefIntervalFromCompletion() TaskDefOption

func WithDefMaxConcurrency added in v1.0.0 

func WithDefMaxConcurrency(n int) TaskDefOption

func WithDefMetadata added in v1.0.0 

func WithDefMetadata(v any) TaskDefOption

func WithDefMetadataRaw added in v1.0.0 

func WithDefMetadataRaw(raw json.RawMessage) TaskDefOption

func WithDefStartAt added in v1.0.0 

func WithDefStartAt(t time.Time) TaskDefOption

func WithDefStartImmediately added in v1.0.0 

func WithDefStartImmediately() TaskDefOption

func WithDefTags added in v1.0.0 

func WithDefTags(tags ...string) TaskDefOption

func WithDefTimeout added in v1.0.0 

func WithDefTimeout(timeout time.Duration) TaskDefOption

func WithDescription added in v1.0.0 

func WithDescription(desc string) TaskDefOption

func WithMaxConcurrency added in v1.0.0 

func WithMaxConcurrency(n int) TaskDefOption

func WithMeta added in v1.0.0 

func WithMeta(v any) TaskDefOption

func WithMetaRaw added in v1.0.0 

func WithMetaRaw(raw json.RawMessage) TaskDefOption

简化命名的 TaskDefOption 别名

func WithStartAt added in v1.0.0 

func WithStartAt(t time.Time) TaskDefOption

func WithTags added in v1.0.0 

func WithTags(tags ...string) TaskDefOption

func WithTimeout added in v1.0.0 

func WithTimeout(d time.Duration) TaskDefOption

type TaskDefinition added in v1.0.0 

type TaskDefinition struct {
	TaskCore
	Metadata json.RawMessage
}

TaskDefinition 描述了一个可调度任务的完整定义（用户输入）

func Cron added in v1.0.0 

func Cron(id, factoryKey, cronSpec string, opts ...TaskDefOption) *TaskDefinition

简化命名的构造函数别名

func Daily added in v1.0.0 

func Daily(id, factoryKey string, atTimes []AtTime, opts ...TaskDefOption) *TaskDefinition

func Every added in v1.0.0 

func Every(id, factoryKey string, interval time.Duration, opts ...TaskDefOption) *TaskDefinition

func Monthly added in v1.0.0 

func Monthly(id, factoryKey string, days []int, atTimes []AtTime, opts ...TaskDefOption) *TaskDefinition

func MonthlyL added in v1.0.0 

func MonthlyL(id, factoryKey string, offset int, atTimes []AtTime, opts ...TaskDefOption) *TaskDefinition

func MonthlyLB added in v1.0.0 

func MonthlyLB(id, factoryKey string, atTimes []AtTime, opts ...TaskDefOption) *TaskDefinition

func MonthlyNthW added in v1.0.0 

func MonthlyNthW(id, factoryKey string, weekday int, nth int, atTimes []AtTime, opts ...TaskDefOption) *TaskDefinition

func NewCronTask added in v1.0.0 

func NewCronTask(id, factoryKey, cronSpec string, opts ...TaskDefOption) *TaskDefinition

func NewDailyTask added in v1.0.0 

func NewDailyTask(id, factoryKey string, atTimes []AtTime, opts ...TaskDefOption) *TaskDefinition

func NewEveryTask added in v1.0.0 

func NewEveryTask(id, factoryKey string, interval time.Duration, opts ...TaskDefOption) *TaskDefinition

func NewMonthlyLastBusinessDayTask added in v1.0.0 

func NewMonthlyLastBusinessDayTask(id, factoryKey string, atTimes []AtTime, opts ...TaskDefOption) *TaskDefinition

func NewMonthlyLastDayTask added in v1.0.0 

func NewMonthlyLastDayTask(id, factoryKey string, offset int, atTimes []AtTime, opts ...TaskDefOption) *TaskDefinition

func NewMonthlyNthWeekdayTask added in v1.0.0 

func NewMonthlyNthWeekdayTask(id, factoryKey string, weekday int, nth int, atTimes []AtTime, opts ...TaskDefOption) *TaskDefinition

func NewMonthlyTask added in v1.0.0 

func NewMonthlyTask(id, factoryKey string, days []int, atTimes []AtTime, opts ...TaskDefOption) *TaskDefinition

func NewOnetimeTask added in v1.0.0 

func NewOnetimeTask(id, factoryKey string, executeAt time.Time, opts ...TaskDefOption) *TaskDefinition

func NewRandomTask added in v1.0.0 

func NewRandomTask(id, factoryKey string, min, max time.Duration, opts ...TaskDefOption) *TaskDefinition

func NewSpecTask added in v1.0.0 

func NewSpecTask(id, factoryKey, spec string, opts ...TaskDefOption) *TaskDefinition

func NewWeeklyTask added in v1.0.0 

func NewWeeklyTask(id, factoryKey string, days []int, atTimes []AtTime, opts ...TaskDefOption) *TaskDefinition

func Onetime added in v1.0.0 

func Onetime(id, factoryKey string, executeAt time.Time, opts ...TaskDefOption) *TaskDefinition

func Random added in v1.0.0 

func Random(id, factoryKey string, min, max time.Duration, opts ...TaskDefOption) *TaskDefinition

func Spec added in v1.0.0 

func Spec(id, factoryKey, spec string, opts ...TaskDefOption) *TaskDefinition

func Weekly added in v1.0.0 

func Weekly(id, factoryKey string, days []int, atTimes []AtTime, opts ...TaskDefOption) *TaskDefinition

type TaskEvent added in v1.0.0 

type TaskEvent struct {
	*TaskDefinition
	Type       TaskEventType `json:"type"`
	Namespace  string        `json:"namespace"`
	InstanceID string        `json:"instance_id"`
	Timestamp  time.Time     `json:"timestamp"` // LastRun 时间（兼容旧版本）
	NextRun    *time.Time    `json:"next_run"`  // NextRun 时间（兼容旧版本）
	Status     *TaskStatus   `json:"status"`    // 完整的任务状态（新版本使用）
}

TaskEvent 任务事件（系统内部事件）

func (*TaskEvent) GetMetadata added in v1.0.0 

func (e *TaskEvent) GetMetadata() json.RawMessage

func (*TaskEvent) GetTaskID added in v1.0.0 

func (e *TaskEvent) GetTaskID() string

type TaskEventType added in v1.0.0 

type TaskEventType string

TaskEventType 任务事件类型 

const (
	TaskEventAdd    TaskEventType = "task_add"
	TaskEventUpdate TaskEventType = "task_update"
	TaskEventRemove TaskEventType = "task_remove"
	TaskEventStart  TaskEventType = "scheduler_start"
	TaskEventStop   TaskEventType = "scheduler_stop"
	TaskEventStatus TaskEventType = "task_status" // 用于同步任务的完整状态（LastRun、NextRun 等）
)

type TaskHandler 

type TaskHandler interface {
	ProcessTask(ctx context.Context, payload []byte) error
}

TaskHandler is the interface for handling tasks (duplicated from consume.go for Redis implementation)

type TaskHandlerFactory added in v1.0.0 

type TaskHandlerFactory func(taskID string, metadata json.RawMessage) (func(context.Context) error, error)

TaskHandlerFactory 任务处理器工厂函数 简化接口：只需要 taskID 和 metadata 即可创建 handler

type TaskMessage added in v1.0.0 

type TaskMessage struct {
	TaskID string `json:"task_id"`
}

TaskMessage 任务消息（用于队列传递）

type TaskOption added in v1.0.0 

type TaskOption func(*taskConfig)

TaskOption 任务操作选项

func WithFactory added in v1.0.0 

func WithFactory(factory TaskHandlerFactory) TaskOption

func WithTaskSilent added in v1.0.0 

func WithTaskSilent(silent bool) TaskOption

func WithTaskTimeout added in v1.0.0 

func WithTaskTimeout(timeout time.Duration) TaskOption

type TaskStatus added in v1.0.0 

type TaskStatus struct {
	ID        string
	Enabled   bool
	LastRun   *time.Time
	NextRun   *time.Time
	RunCount  int64
	IsRunning bool
	// IsLeader 表示当前节点是否为 Leader（仅在分布式模式下有意义）
	// 非 Leader 节点无法获取 LastRun 和 NextRun 信息
	IsLeader bool
}

TaskStatus 任务状态信息

type WeeklySchedule added in v1.0.0 

type WeeklySchedule struct {
	Interval   int
	DaysOfWeek []int
	AtTimes    []AtTime
}

WeeklySchedule 每周调度配置

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.