生产环境go-redsync使用示例

package main

import (
	goredislib "github.com/redis/go-redis/v9"
	"github.com/go-redsync/redsync/v4"
	"github.com/go-redsync/redsync/v4/redis/goredis/v9"
)

func main() {
	// Create a pool with go-redis (or redigo) which is the pool redisync will
	// use while communicating with Redis. This can also be any pool that
	// implements the `redis.Pool` interface.
	client := goredislib.NewClient(&goredislib.Options{
		Addr: "localhost:6379",
	})
	pool := goredis.NewPool(client) // or, pool := redigo.NewPool(...)

	// Create an instance of redisync to be used to obtain a mutual exclusion
	// lock.
	rs := redsync.New(pool)

	// Obtain a new mutex by using the same name for all instances wanting the
	// same lock.
	mutexname := "my-global-mutex"
	mutex := rs.NewMutex(mutexname)

	// Obtain a lock for our given mutex. After this is successful, no one else
	// can obtain the same lock (the same mutex name) until we unlock it.
	if err := mutex.Lock(); err != nil {
		panic(err)
	}

	// Do your work that requires the lock.

	// Release the lock so other processes or threads can obtain a lock.
	if ok, err := mutex.Unlock(); !ok || err != nil {
		panic("unlock failed")
	}
}

// 这里省去创建redis连接的操作

mutexname := "my-global-mutex"
var wg sync.WaitGroup // 用于实现主函数等待所有子协程执行完毕之后再退出
wg.Add(2)
go func() {
   defer wg.Done()
   mutex := rs.NewMutex(mutexname)
   // 开始尝试获得分布式锁
   if err := mutex.Lock(); err != nil {
      log.Errorf("failed to acquire lock in task1: %v", err)
      return
   }

   // 执行一些任务
   log.Info("task1 start at ", time.Now().Format("15:04:05.000"))
   time.Sleep(time.Second * 10) // 模拟一个耗时的任务
   log.Info("task1 end at ", time.Now().Format("15:04:05.000"))

   // 执行完任务，释放锁
   if _, err := mutex.Unlock(); err != nil {
      log.Errorf("failed to release lock in task1: %v", err)
   }
}()
go func() {
   defer wg.Done()
   mutex := rs.NewMutex(mutexname)
   // 开始尝试获得分布式锁
   if err := mutex.Lock(); err != nil {
      log.Errorf("failed to acquire lock in task2: %v", err)
      return
   }

   // 执行一些任务
   log.Info("task2 start at ", time.Now().Format("15:04:05.000"))
   time.Sleep(time.Second * 10) // 模拟一个耗时的任务
   log.Info("task2 end at ", time.Now().Format("15:04:05.000"))

   // 执行完任务，释放锁
   if _, err := mutex.Unlock(); err != nil {
      log.Errorf("failed to release lock in task2: %v", err)
   }
}()
wg.Wait()

mutexname := "my-global-mutex"
var wg sync.WaitGroup // 用于实现主函数等待所有子协程执行完毕之后再退出
wg.Add(2)
go func() {
   defer wg.Done()
   mutex := client.NewMutex(mutexname)
   // 开始尝试获得分布式锁
   if err := mutex.Lock(); err != nil {
      log.Errorf("failed to acquire lock in task1: %v", err)
      return
   }
   var lockReleased atomic.Bool
   lockReleased.Store(false)

   go func() { // 只要当前任务还在执行，每过1秒就延长锁的过期时间
      for {
         time.Sleep(time.Second)
         if lockReleased.Load() {
            return
         }
         _, err := mutex.Extend()
         if err != nil {
            log.Errorf("extend lock in task1 fail: %v", err)
         }
      }
   }()

   log.Info("task1 start at ", time.Now().Format("15:04:05.000"))
   time.Sleep(time.Second * 10) // 模拟一个耗时的任务
   log.Info("task1 end at ", time.Now().Format("15:04:05.000"))

   // 执行完任务，释放锁
   if _, err := mutex.Unlock(); err != nil {
      log.Errorf("failed to release lock in task1: %v", err)
   }
   lockReleased.Store(true)
}()
go func() {
   defer wg.Done()
   mutex := client.NewMutex(mutexname)
   // 开始尝试获得分布式锁
   if err := mutex.Lock(); err != nil {
      log.Errorf("failed to acquire lock in task2: %v", err)
      return
   }
   var lockReleased atomic.Bool
   lockReleased.Store(false)

   go func() { // 只要当前任务还在执行，每过1秒就延长锁的过期时间
      for {
         time.Sleep(time.Second)
         if lockReleased.Load() {
            return
         }
         _, err := mutex.Extend()
         if err != nil {
            log.Errorf("extend lock in task2 fail: %v", err)
         }
      }
   }()

   log.Info("task2 start at ", time.Now().Format("15:04:05.000"))
   time.Sleep(time.Second * 10) // 模拟一个耗时的任务
   log.Info("task2 end at ", time.Now().Format("15:04:05.000"))

   // 执行完任务，释放锁
   if _, err := mutex.Unlock(); err != nil {
      log.Errorf("failed to release lock in task2: %v", err)
   }
   lockReleased.Store(true)
}()
wg.Wait()

type LockHolder interface {
    ReleaseLock()
}

type lockHolder struct {
    mutex       *redsync.Mutex
    lockReleased atomic.Bool
}

func (h *lockHolder) ReleaseLock() {
    _, err := h.mutex.Unlock()
    if err != nil {
        log.Errorf("failed to release lock: %v", err)
    }
    h.lockReleased.Store(true)
}

func TryLock(mutexName string) (LockHolder, error) {
    rs := getRedsync()
    mutex := rs.NewMutex(mutexName)
    if err := mutex.Lock(); err != nil {
        log.Errorf("failed to acquire lock: %v", err)
        return nil, err
    }

    holder := &lockHolder{
        mutex:       mutex,
        lockReleased: atomic.Bool{},
    }
    holder.lockReleased.Store(false)

    go func() {
        for {
            time.Sleep(time.Second)
            if holder.lockReleased.Load() {
                return
            }
            _, err := mutex.Extend()
            if err != nil {
                log.Errorf("extend lock fail: %v", err)
            }
        }
    }()
    return holder, nil
}

mutexname := "my-global-mutex"
var wg sync.WaitGroup
wg.Add(2)
go func() {
   defer wg.Done()
   lock, err := TryLock(mutexname)
   if err != nil {
      log.Errorf("failed to acquire lock in task1: %v", err)
      return
   }
   defer lock.ReleaseLock()
   log.Info("task1 start at ", time.Now().Format("15:04:05.000"))
   time.Sleep(time.Second * 10)
   log.Info("task1 end at ", time.Now().Format("15:04:05.000"))
}()
go func() {
   defer wg.Done()
   lock, err := TryLock(mutexname)
   if err != nil {
      log.Errorf("failed to acquire lock in task2: %v", err)
      return
   }
   defer lock.ReleaseLock()
   log.Info("task2 start at ", time.Now().Format("15:04:05.000"))
   time.Sleep(time.Second * 10)
   log.Info("task2 end at ", time.Now().Format("15:04:05.000"))
}()
wg.Wait()