cydist

package
v0.0.1 Latest Latest
Warning

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

 Go to latest Published: Sep 23, 2025 License: MIT Imports: 19 Imported by: 0

README

MemoryStore - 基于 Redis 的内存记录系统

MemoryStore 是一个基于 Redis 的内存记录系统，允许您根据某个 key 记录多条记录，并提供以下功能：

  1. 为每个 key 设置整体超时时间
  2. 重新写入时重新计算超期时间
  3. 获取最新的多条数据
  4. 为每个 key 设置记录上限，超过的记录会被丢弃
  5. 支持命名空间隔离，确保多个服务间的数据不会相互干扰
  6. 配置信息持久化存储在 Redis 中，服务重启后配置不会丢失
  7. 本地缓存机制，提高配置获取性能

功能特性

  • 配置管理：可以预先为 key 设置配置（超时时间和最大记录数），配置信息持久化存储在 Redis 中
  • 记录存储：支持向指定 key 添加记录
  • 记录获取：支持获取指定 key 的最新记录
  • 记录限制：自动维护每个 key 的最大记录数
  • 自动过期：支持为每个 key 设置过期时间
  • 命名空间隔离：支持通过命名空间隔离不同服务的数据
  • 持久化配置：配置信息存储在 Redis 中，服务重启后不会丢失
  • 本地缓存：配置信息在本地缓存，提高获取性能，默认缓存5分钟

安装和使用

1. 创建 MemoryStore 实例
import "github.com/fj1981/infrakit/pkg/cydist"

// 创建 Redis 客户端
client := cydist.New(cydist.WithAddr("localhost:6379"))

// 创建 MemoryStore 实例，指定命名空间
ms := cydist.NewMemoryStore(client, "myservice")
2. 配置缓存设置（可选）
// 设置配置缓存的过期时间
ms.SetConfigCacheTTL(10 * time.Minute)
3. 预先配置 key（推荐方式）
// 为 key 配置超时时间和最大记录数
key := "user_actions"
expire := 10 * time.Minute
maxRecords := int64(100)

err := ms.CreateKey(context.Background(), key, expire, maxRecords)
if err != nil {
    // 处理错误
}

// 添加记录（无需每次都指定配置）
data := map[string]interface{}{
    "user_id": 12345,
    "action": "login",
    "timestamp": time.Now(),
}

err = ms.WriteRecord(context.Background(), key, data)
if err != nil {
    // 处理错误
}
4. 直接写入记录（向后兼容方式）
// 也可以直接指定配置参数（向后兼容）
key := "user_actions"
data := map[string]interface{}{
    "user_id": 12345,
    "action": "login",
    "timestamp": time.Now(),
}
expire := 10 * time.Minute
maxRecords := int64(100)

err := ms.WriteRecordWithConfig(context.Background(), key, data, expire, maxRecords)
if err != nil {
    // 处理错误
}
5. 获取记录
// 获取最新的 10 条记录
records, err := ms.GetLatestRecords(context.Background(), key, 10)
if err != nil {
    // 处理错误
}

for _, record := range records {
    fmt.Printf("Data: %v, Timestamp: %v\n", record.Data, record.Timestamp)
}
6. 删除记录
// 删除指定 key 的所有记录和配置
deleted, err := ms.DeleteRecords(context.Background(), key)
if err != nil {
    // 处理错误
}

API 参考

type MemoryStore
type MemoryStore struct {
    // contains filtered or unexported fields
}
func NewMemoryStore
func NewMemoryStore(client *RedisClient, namespace string) *MemoryStore

NewMemoryStore 创建一个新的 MemoryStore 实例，namespace 参数用于隔离不同服务的数据。

func (*MemoryStore) SetConfigCacheTTL
func (ms *MemoryStore) SetConfigCacheTTL(ttl time.Duration)

SetConfigCacheTTL 设置配置缓存的过期时间，默认为5分钟。

func (*MemoryStore) CreateKey
func (ms *MemoryStore) CreateKey(ctx context.Context, key string, expire time.Duration, maxRecords int64) error

CreateKey 为指定的 key 创建配置，包括过期时间和最大记录数。配置信息会持久化存储在 Redis 中，并在本地缓存以提高性能。

func (*MemoryStore) DeleteKey
func (ms *MemoryStore) DeleteKey(ctx context.Context, key string) (int64, error)

DeleteKey 删除指定 key 的配置，并从本地缓存中移除。

func (*MemoryStore) WriteRecord
func (ms *MemoryStore) WriteRecord(ctx context.Context, key string, data interface{}) error

WriteRecord 向预先配置的 key 添加一条记录。配置信息会优先从本地缓存获取，如果缓存未命中或已过期，则从 Redis 中获取。

func (*MemoryStore) WriteRecordWithConfig
func (ms *MemoryStore) WriteRecordWithConfig(ctx context.Context, key string, data interface{}, expire time.Duration, maxRecords int64) error

WriteRecordWithConfig 向指定 key 添加一条记录，并指定配置（向后兼容方法）。

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

GetLatestRecords 获取指定 key 的最新记录。

func (*MemoryStore) GetRecordCount
func (ms *MemoryStore) GetRecordCount(ctx context.Context, key string) (int64, error)

GetRecordCount 获取指定 key 的记录数量。

func (*MemoryStore) DeleteRecords
func (ms *MemoryStore) DeleteRecords(ctx context.Context, key string) (int64, error)

DeleteRecords 删除指定 key 的所有记录和配置。

type Record
type Record struct {
    Data      interface{} `json:"data"`
    Timestamp time.Time   `json:"timestamp"`
}

Record 表示一条记录。

type KeyNotConfiguredError
type KeyNotConfiguredError struct {
    Key string
}

KeyNotConfiguredError 表示 key 未配置的错误。

缓存机制说明

MemoryStore 使用本地缓存来提高配置获取的性能：

  1. 当调用 WriteRecord 时，首先尝试从本地缓存获取配置信息
  2. 如果缓存未命中或已过期，则从 Redis 中获取配置信息，并更新本地缓存
  3. 默认缓存时间为5分钟，可以通过 SetConfigCacheTTL 方法调整
  4. MemoryStore 会定期清理过期的缓存项

注意事项

  1. 使用前请确保 Redis 服务正在运行
  2. 推荐使用 CreateKey 和 WriteRecord 方法，这样不需要每次都指定配置参数
  3. 当使用 WriteRecordWithConfig 方法时，配置仅在该次调用中有效
  4. 记录会自动按时间倒序存储（最新的记录在最前面）
  5. 当记录数超过限制时，旧的记录会被自动删除
  6. 使用命名空间可以有效隔离不同服务的数据，避免相互干扰
  7. 配置信息持久化存储在 Redis 中，服务重启后配置不会丢失
  8. 本地缓存机制可以显著提高配置获取性能，减少 Redis 访问次数

Documentation

Overview

distlock.go

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

Index

Constants

This section is empty.

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 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 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) 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 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 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 DistributedScheduler 

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

DistributedScheduler 分布式调度器

func NewDistributedScheduler 

func NewDistributedScheduler(locker *DistLockManager, instanceID string) *DistributedScheduler

NewDistributedScheduler 创建调度器 注意：DistLockManager 应由外部创建并注入

func (*DistributedScheduler) AddTask 

func (s *DistributedScheduler) AddTask(task *Task) (cron.EntryID, error)

AddTask 添加一个分布式任务

func (*DistributedScheduler) ListTasks 

func (s *DistributedScheduler) ListTasks() map[string]string

ListTasks 获取所有任务

func (*DistributedScheduler) RemoveTask 

func (s *DistributedScheduler) RemoveTask(taskID string) bool

RemoveTask 删除任务

func (*DistributedScheduler) Start 

func (s *DistributedScheduler) Start()

Start 启动调度器

func (*DistributedScheduler) Stop 

func (s *DistributedScheduler) Stop()

Stop 停止调度器

func (*DistributedScheduler) StopRunningTask 

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

StopRunningTask 取消正在运行的任务

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 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 WithLockTimeout 

func WithLockTimeout(timeout time.Duration) LockOption

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

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 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 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) 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) 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) 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) 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 Task 

type Task struct {
	ID      string                      // 任务唯一ID
	Spec    string                      // Cron 表达式，如 "0 2 * * *"
	Handler func(context.Context) error // 任务执行函数
	Timeout time.Duration               // 执行超时时间
}

Task 定义一个可调度的任务

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)

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.