生產(chǎn)環(huán)境go-redsync使用示例

更新時(shí)間：2025年10月01日 11:34:30 作者：qinyuan15

go-redsync是go語(yǔ)言實(shí)現(xiàn)分布式鎖的常用工具,但官方文檔是的入門示例并不是一個(gè)可以直接用于生產(chǎn)環(huán)境的版本,本文提供一個(gè)可以用于生產(chǎn)環(huán)境的使用示例,感興趣的可以了解一下

一、問(wèn)題和意義

go-redsync是go語(yǔ)言實(shí)現(xiàn)分布式鎖的常用工具，但官方文檔是的入門示例并不是一個(gè)可以直接用于生產(chǎn)環(huán)境的版本。很多人將官方文檔中的入門示例使用到實(shí)際項(xiàng)目中導(dǎo)致了生產(chǎn)事故。故文本提供一個(gè)可以用于生產(chǎn)環(huán)境的使用示例。

二、官方入門示例存在的問(wèn)題

官方示例代碼為：

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")
	}
}

接下來(lái)，我們開兩個(gè)協(xié)程實(shí)測(cè)一下。

// 這里省去創(chuàng)建redis連接的操作

mutexname := "my-global-mutex"
var wg sync.WaitGroup // 用于實(shí)現(xiàn)主函數(shù)等待所有子協(xié)程執(zhí)行完畢之后再退出
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
   }

   // 執(zhí)行一些任務(wù)
   log.Info("task1 start at ", time.Now().Format("15:04:05.000"))
   time.Sleep(time.Second * 10) // 模擬一個(gè)耗時(shí)的任務(wù)
   log.Info("task1 end at ", time.Now().Format("15:04:05.000"))

   // 執(zhí)行完任務(wù)，釋放鎖
   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
   }

   // 執(zhí)行一些任務(wù)
   log.Info("task2 start at ", time.Now().Format("15:04:05.000"))
   time.Sleep(time.Second * 10) // 模擬一個(gè)耗時(shí)的任務(wù)
   log.Info("task2 end at ", time.Now().Format("15:04:05.000"))

   // 執(zhí)行完任務(wù)，釋放鎖
   if _, err := mutex.Unlock(); err != nil {
      log.Errorf("failed to release lock in task2: %v", err)
   }
}()
wg.Wait()

程序執(zhí)行結(jié)果如下：

INFO msg=task2 start at 02:22:00.330
INFO msg=task1 start at 02:22:08.508
INFO msg=task2 end at 02:22:10.330
ERROR msg=failed to release lock in task2: lock already taken, locked nodes: [0]
INFO msg=task1 end at 02:22:18.508
ERROR msg=failed to release lock in task1: lock already taken, locked nodes: [0]

可以看出，分布式鎖并沒(méi)有起作用，任務(wù)2還沒(méi)執(zhí)行完，任務(wù)1就已經(jīng)獲得鎖并開始。原因是go-redsync默認(rèn)的加鎖時(shí)間只有8秒鐘，如果一個(gè)任務(wù)執(zhí)行時(shí)間超過(guò)8秒，則分布式鎖會(huì)在任務(wù)執(zhí)行結(jié)束前釋放。

三、生產(chǎn)環(huán)境可用的版本

生產(chǎn)環(huán)境中，任務(wù)沒(méi)結(jié)束時(shí)需要調(diào)用mutex.Extend()方法延長(zhǎng)鎖的時(shí)間

mutexname := "my-global-mutex"
var wg sync.WaitGroup // 用于實(shí)現(xiàn)主函數(shù)等待所有子協(xié)程執(zhí)行完畢之后再退出
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() { // 只要當(dāng)前任務(wù)還在執(zhí)行，每過(guò)1秒就延長(zhǎng)鎖的過(guò)期時(shí)間
      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) // 模擬一個(gè)耗時(shí)的任務(wù)
   log.Info("task1 end at ", time.Now().Format("15:04:05.000"))

   // 執(zhí)行完任務(wù)，釋放鎖
   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() { // 只要當(dāng)前任務(wù)還在執(zhí)行，每過(guò)1秒就延長(zhǎng)鎖的過(guò)期時(shí)間
      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) // 模擬一個(gè)耗時(shí)的任務(wù)
   log.Info("task2 end at ", time.Now().Format("15:04:05.000"))

   // 執(zhí)行完任務(wù)，釋放鎖
   if _, err := mutex.Unlock(); err != nil {
      log.Errorf("failed to release lock in task2: %v", err)
   }
   lockReleased.Store(true)
}()
wg.Wait()

執(zhí)行結(jié)果如下：

INFO msg=task2 start at 02:31:06.973
INFO msg=task2 end at 02:31:16.974
INFO msg=task1 start at 02:31:17.471
INFO msg=task1 end at 02:31:27.471

這一執(zhí)行結(jié)果符合預(yù)期，任務(wù)1會(huì)在任務(wù)2執(zhí)行完之后才能獲得鎖。

四、對(duì)go-redsync做封裝

前面的代碼示例可用于生產(chǎn)，但代碼過(guò)于冗長(zhǎng)，每次使用分布式鎖時(shí)都寫那么多代碼也太麻煩。我們可以將其封裝為了個(gè)TryLock方法：

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
}

接下來(lái)使用分布式鎖的代碼可以簡(jiǎn)化為：

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

到此這篇關(guān)于生產(chǎn)環(huán)境go-redsync使用示例的文章就介紹到這了,更多相關(guān)生產(chǎn)環(huán)境go-redsync使用內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: