Documentation ¶
Index ¶
Examples ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
Types ¶
type Conn ¶
type Conn interface { // SetNX 存储锁的key和value, 如果需要设置连接超时,可以使用context.WithTimeout, // 如果超时时间比过期时间(expires)大,使用expires作为超时时间 SetNX(ctx context.Context, key, value string, expires time.Duration) error // Delete 删除锁的key, 如果需要设置连接超时,可以使用context.WithTimeout Delete(ctx context.Context, key, value string) error // Extend 存储锁的key的延长过期时间, 如果需要设置连接超时,可以使用context.WithTimeout, // 如果超时时间比过期时间(expires)大,使用expires作为超时时间 Extend(ctx context.Context, key, value string, expires time.Duration) error // Close 关闭redis的连接 Close() error }
Conn redis连接
type MultiError ¶
type MultiError []error
MultiError 包装多个错误
func (MultiError) Unwrap ¶
func (m MultiError) Unwrap() error
type Mutex ¶
type Mutex struct {
// contains filtered or unexported fields
}
Mutex 互斥锁
func (*Mutex) Exec ¶
func (m *Mutex) Exec(ctx context.Context, f WorkerFunc) (err error)
Exec 加锁并执行 传入的f函数如果执行时间过长, 会尝试延续锁的有效期, 延长有效期失败就退出,至于执行的操作是否有中断方法需要在f函数内部使用context
Example ¶
package main import ( "context" "fmt" "time" "gitee.com/wqt/redlock" "gitee.com/wqt/redlock/driver/redigo" "github.com/gomodule/redigo/redis" ) func main() { redlock.SetLogger(&redlock.StdLogger{}) // 初始化redigo的连接 p := &redis.Pool{ DialContext: func(ctx context.Context) (redis.Conn, error) { return redis.DialContext(ctx, "tcp", "127.0.0.1:6379", redis.DialDatabase(2), ) }, MaxIdle: 2, IdleTimeout: 2 * time.Minute, TestOnBorrow: func(c redis.Conn, t time.Time) error { if time.Since(t) < time.Minute { return nil } _, err := c.Do("PING") return err }, } taskName := func(id string) string { return fmt.Sprintf("task:%s", id) } // 创建一个工作任务锁 workers := redlock.New(redigo.NewClient(p)) lock := workers.NewMutex(taskName("exec"), // 请求超时时间 redlock.WithTimeout(30*time.Second), // 锁的有效期 redlock.WithExpires(60*time.Second), // 重试次数, 因此最大尝试次数是2+1 redlock.WithRetries(2), ) ctx, cancel := context.WithCancel(context.Background()) defer cancel() if err := lock.Exec(ctx, func(c context.Context) { select { case <-ctx.Done(): return default: } fmt.Println("ok") }); err != nil { fmt.Println(err.Error()) } }
Output: ok
Example (ExtendCanceled) ¶
package main import ( "context" "fmt" "time" "gitee.com/wqt/redlock" "gitee.com/wqt/redlock/driver/redigo" "github.com/gomodule/redigo/redis" ) func main() { // 初始化redigo的连接 p := &redis.Pool{ DialContext: func(ctx context.Context) (redis.Conn, error) { return redis.DialContext(ctx, "tcp", "127.0.0.1:6379", redis.DialDatabase(2), ) }, MaxIdle: 2, IdleTimeout: 2 * time.Minute, } taskName := func(id string) string { return fmt.Sprintf("task:%s", id) } // 创建一个工作任务锁 workers := redlock.New(redigo.NewClient(p)) lock := workers.NewMutex(taskName("exec:extend:canceled"), // 锁的有效期 redlock.WithExpires(500*time.Millisecond), // 重试次数, 因此最大尝试次数是2+1 redlock.WithRetries(2), ) ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) defer cancel() exit := make(chan struct{}) err := lock.Exec(ctx, func(c context.Context) { defer close(exit) for i := 0; i < 30; i++ { select { case <-ctx.Done(): fmt.Println("cancel") return default: time.Sleep(10 * time.Millisecond) } } fmt.Println("ok") }) if err != nil { fmt.Println("err =", err.Error()) } select { case <-exit: case <-ctx.Done(): time.Sleep(10 * time.Millisecond) } }
Output: err = context deadline exceeded cancel
Example (ExtendOk) ¶
package main import ( "context" "fmt" "time" "gitee.com/wqt/redlock" "gitee.com/wqt/redlock/driver/redigo" "github.com/gomodule/redigo/redis" ) func main() { // 初始化redigo的连接 redlock.SetLogger(&redlock.StdLogger{}) p := &redis.Pool{ DialContext: func(ctx context.Context) (redis.Conn, error) { return redis.DialContext(ctx, "tcp", "127.0.0.1:6379", redis.DialDatabase(2), ) }, MaxIdle: 2, IdleTimeout: 2 * time.Minute, } taskName := func(id string) string { return fmt.Sprintf("task:%s", id) } // 创建一个工作任务锁 workers := redlock.New(redigo.NewClient(p)) lock := workers.NewMutex(taskName("exec:extend:ok"), // 请求超时时间 redlock.WithTimeout(2*time.Second), // 锁的有效期 redlock.WithExpires(500*time.Millisecond), // 重试次数, 因此最大尝试次数是2+1 redlock.WithRetries(2), ) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() exit := make(chan struct{}) err := lock.Exec(ctx, func(ctx context.Context) { defer close(exit) for i := 0; i < 2; i++ { select { case <-ctx.Done(): fmt.Println("cancel") default: time.Sleep(200 * time.Millisecond) } } fmt.Println("ok") }) if err != nil { fmt.Println("err =", err.Error()) } select { case <-exit: case <-ctx.Done(): time.Sleep(10 * time.Millisecond) } }
Output: ok
type Option ¶
type Option func(*Options)
Option 选项函数
func WithConnections ¶
WithConnections 设置客户端连接, 并不检查是否是nil, 传入的数量就是所有的redis实例数
type WorkerFunc ¶
WorkerFunc 工作函数, 任务可以选择接收ctx.Done() 由更上层的上下文取消时, 执行退出操作, 也可以忽略此取消事件;
Click to show internal directories.
Click to hide internal directories.