refactor: 将分布式锁从 redis 迁移至 utils 包
This commit is contained in:
@@ -5,6 +5,7 @@ import (
|
||||
"fmt"
|
||||
"net/http"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"gitea.com/red-future/common/utils"
|
||||
@@ -53,11 +54,11 @@ func init() {
|
||||
}
|
||||
}
|
||||
|
||||
func UploadFile(ctx context.Context, fileHeader *ghttp.UploadFile) (imagesUrl string, err error) {
|
||||
func UploadFile(ctx context.Context, fileHeader *ghttp.UploadFile) (imagesUrl string, fileName string, fileFormat string, err error) {
|
||||
return uploadFile(ctx, fileHeader)
|
||||
}
|
||||
|
||||
func uploadFile(ctx context.Context, fileHeader *ghttp.UploadFile) (imagesUrl string, err error) {
|
||||
func uploadFile(ctx context.Context, fileHeader *ghttp.UploadFile) (imagesUrl string, fileName string, fileFormat string, err error) {
|
||||
bucketName, err := utils.GetBucketName(ctx)
|
||||
if err != nil {
|
||||
glog.Errorf(ctx, "获取桶名称失败: %v", err)
|
||||
@@ -124,5 +125,5 @@ func uploadFile(ctx context.Context, fileHeader *ghttp.UploadFile) (imagesUrl st
|
||||
glog.Errorf(ctx, "上传图片失败: %v", err)
|
||||
return
|
||||
}
|
||||
return objectName, err
|
||||
return objectName, fileHeader.Filename, strings.ReplaceAll(fileExt, ".", ""), err
|
||||
}
|
||||
|
||||
@@ -2,7 +2,6 @@ package redis
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
@@ -38,40 +37,6 @@ func RedisClient() *gredis.Redis {
|
||||
return getClient()
|
||||
}
|
||||
|
||||
// Lock 分布式锁
|
||||
func Lock(ctx context.Context, key string, expireSeconds int64, fn func(ctx context.Context) error) (success bool, err error) {
|
||||
limit := 3
|
||||
LOOP:
|
||||
if limit < 0 {
|
||||
return false, errors.New("锁重试次数耗尽")
|
||||
}
|
||||
limit--
|
||||
client := getClient()
|
||||
if val, err := client.Set(ctx, key, true, gredis.SetOption{
|
||||
TTLOption: gredis.TTLOption{
|
||||
EX: &expireSeconds,
|
||||
},
|
||||
NX: true,
|
||||
}); err != nil {
|
||||
return false, err
|
||||
} else {
|
||||
if val.Bool() {
|
||||
defer func(client *gredis.Redis, ctx context.Context, key string) {
|
||||
if _, err = client.Del(ctx, key); err != nil {
|
||||
glog.Errorf(ctx, "redis client Del error: %v", err)
|
||||
}
|
||||
}(client, ctx, key)
|
||||
if err = fn(ctx); err != nil {
|
||||
return false, err
|
||||
}
|
||||
return true, nil
|
||||
} else {
|
||||
time.Sleep(time.Second)
|
||||
goto LOOP
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func GetReadStream(ctx context.Context, msg ...QueueMessage) error {
|
||||
for _, t := range msg {
|
||||
err := GetReadFromStream(ctx, t.StreamKey, t.GroupName, t.ConsumerName, t.BatchSize, t.BlockMs, t.AutoAck, t.HandleFunc)
|
||||
|
||||
@@ -3,6 +3,7 @@ package utils
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"reflect"
|
||||
@@ -13,10 +14,12 @@ import (
|
||||
"time"
|
||||
|
||||
"gitea.com/red-future/common/beans"
|
||||
"github.com/gogf/gf/v2/container/gvar"
|
||||
"github.com/gogf/gf/v2/database/gredis"
|
||||
"github.com/gogf/gf/v2/errors/gcode"
|
||||
"github.com/gogf/gf/v2/errors/gerror"
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
"github.com/gogf/gf/v2/os/glog"
|
||||
"github.com/gogf/gf/v2/os/gtime"
|
||||
"github.com/gogf/gf/v2/util/gconv"
|
||||
"github.com/tiger1103/gfast-token/gftoken"
|
||||
@@ -397,6 +400,7 @@ func GetFileAddressPrefix(ctx context.Context) (imageUrl string, err error) {
|
||||
return
|
||||
}
|
||||
|
||||
// GetBucketName 获取bucket名称
|
||||
func GetBucketName(ctx context.Context) (bucketName string, err error) {
|
||||
user, err := GetUserInfo(ctx)
|
||||
if err != nil {
|
||||
@@ -405,3 +409,37 @@ func GetBucketName(ctx context.Context) (bucketName string, err error) {
|
||||
bucketName = fmt.Sprintf("tenantid-%d", user.TenantId)
|
||||
return
|
||||
}
|
||||
|
||||
// Lock 分布式锁
|
||||
func Lock(ctx context.Context, key string, expireSeconds int64, fn func(ctx context.Context) error) (success bool, err error) {
|
||||
limit := 3
|
||||
LOOP:
|
||||
if limit < 0 {
|
||||
return false, errors.New("锁重试次数耗尽")
|
||||
}
|
||||
limit--
|
||||
var val *gvar.Var
|
||||
if val, err = g.Redis().Set(ctx, key, true, gredis.SetOption{
|
||||
TTLOption: gredis.TTLOption{
|
||||
EX: &expireSeconds,
|
||||
},
|
||||
NX: true,
|
||||
}); err != nil {
|
||||
return false, err
|
||||
}
|
||||
|
||||
if val.Bool() {
|
||||
defer func(ctx context.Context, key string) {
|
||||
if _, err = g.Redis().Del(ctx, key); err != nil {
|
||||
glog.Errorf(ctx, "redis client Del error: %v", err)
|
||||
}
|
||||
}(ctx, key)
|
||||
if err = fn(ctx); err != nil {
|
||||
return false, err
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
|
||||
time.Sleep(time.Second)
|
||||
goto LOOP
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user