sorted

bloom/bloom.go

@@ -5,7 +5,6 @@ import (
 	"math"
 
 	"github.com/charlienet/go-mixed/bytesconv"
-	"github.com/charlienet/go-mixed/expr"
 	"github.com/charlienet/go-mixed/hash"
 	"github.com/charlienet/go-mixed/redis"
 )
@@ -58,10 +57,7 @@ func New(expectedInsertions uint, fpp float64, opts ...option) *BloomFilter {
 	bf := &BloomFilter{
 		bits:  bits,
 		funcs: k,
-		store: expr.Ternary[bitStore](
-			opt.redisClient == nil,
-			newMemStore(bits),
-			newRedisStore(opt.redisClient, opt.redisKey, bits)),
+		store: createBitStore(opt, bits),
 	}
 
 	return bf
@@ -104,6 +100,14 @@ func (bf *BloomFilter) Clear() {
 	bf.store.Clear()
 }
 
+func createBitStore(opt *bloomOptions, bits uint) bitStore {
+	if opt.redisClient != nil {
+		return newRedisStore(opt.redisClient, opt.redisKey, bits)
+	}
+
+	return newMemStore(bits)
+}
+
 // 计算优化的位图长度，
 // n 期望放置元素数量，
 // p 预期的误判概率

bloom/mem_store.go

@@ -2,22 +2,21 @@ package bloom
 
 import (
 	"context"
+	"sync"
 
 	"github.com/bits-and-blooms/bitset"
-	"github.com/charlienet/go-mixed/locker"
 )
 
 type memStore struct {
 	size uint
-	set  *bitset.BitSet  // 内存位图
-	lock locker.RWLocker // 同步锁
+	set  *bitset.BitSet // 内存位图
+	lock sync.RWMutex   // 同步锁
 }
 
 func newMemStore(size uint) *memStore {
 	return &memStore{
 		size: size,
 		set:  bitset.New(size),
-		lock: locker.NewRWLocker(),
 	}
 }
 

configure/config.toml

@@ -0,0 +1,7 @@
+Mode = "dev"
+
+[Nacos]
+    Address = "192.168.2.121"
+    Port = 8848
+    Namespace = "8560b58d-87d0-4b85-8ac5-f2d308c6669e"
+    Group = "dev"

configure/configure.go

@@ -0,0 +1,46 @@
+package configure
+
+import (
+	"github.com/charlienet/go-mixed/expr"
+	"github.com/spf13/viper"
+)
+
+type NotifyFunc func(Configure) error
+
+type Configure interface {
+	Load(dataId string, v any, onChanged ...NotifyFunc) error
+	GetString(string, string) string
+	GetInt(key string, defaultValue int) int
+}
+
+type conf struct {
+	viper            *viper.Viper            //
+	nacos            *nacos                  //
+	onChangeNotifies map[string][]NotifyFunc // 已经注册的配置变更通知
+	nacosOptions     *NacosOptions           //
+	useNacos         bool                    //
+}
+
+func (c *conf) GetString(key string, defaultValue string) string {
+	if c.viper.IsSet(key) {
+		return c.viper.GetString(key)
+	}
+
+	return defaultValue
+}
+
+func (c *conf) GetInt(key string, defaultValue int) int {
+	return expr.Ternary(c.viper.IsSet(key), c.viper.GetInt(key), defaultValue)
+}
+
+func (c *conf) Load(dataId string, v any, onChanged ...NotifyFunc) error {
+	if err := c.nacos.Load(dataId, v); err != nil {
+		return err
+	}
+
+	if len(onChanged) > 0 {
+		c.onChangeNotifies[dataId] = onChanged
+	}
+
+	return nil
+}

configure/configure_builder.go

@@ -0,0 +1,100 @@
+package configure
+
+import (
+	"github.com/fsnotify/fsnotify"
+	"github.com/spf13/viper"
+)
+
+func New() *conf { return &conf{viper: viper.New(), onChangeNotifies: make(map[string][]NotifyFunc)} }
+
+func (c *conf) AddConfigPath(in ...string) *conf {
+	for _, v := range in {
+		c.viper.AddConfigPath(v)
+	}
+
+	return c
+}
+
+func (c *conf) SetConfigName(in string) *conf {
+	c.viper.SetConfigName(in)
+	return c
+}
+
+func (c *conf) SetConfigFile(f string) *conf {
+	c.viper.SetConfigFile(f)
+	return c
+}
+
+func (c *conf) SetDefault(key string, value any) *conf {
+	c.viper.SetDefault(key, value)
+	return c
+}
+
+func (c *conf) AutomaticEnv() *conf {
+	c.viper.AutomaticEnv()
+	return c
+}
+
+func (c *conf) Read() (*conf, error) {
+	// 从本地配置读取
+	if err := c.viper.ReadInConfig(); err != nil {
+		return nil, err
+	}
+
+	c.viper.WatchConfig()
+	c.viper.OnConfigChange(c.OnViperChanged)
+
+	// 初始化Nacos客户端
+	if err := c.createNacosClient(); err != nil {
+		return nil, err
+	}
+
+	return c, nil
+}
+
+func (c *conf) OnViperChanged(in fsnotify.Event) {
+
+}
+
+func (c *conf) createNacosClient() error {
+	opt := c.getNacosOptions()
+	if opt == nil {
+		return nil
+	}
+
+	nc, err := createNacosClient(opt.Address, opt.Port, opt.Namespace, opt.Group)
+	if err != nil {
+		return err
+	}
+
+	c.nacos = &nacos{client: nc, group: opt.Group, onChanged: c.onNacosChanged}
+
+	return nil
+}
+
+func (c *conf) onNacosChanged(dataId, data string) {
+	if fs, ok := c.onChangeNotifies[dataId]; ok {
+		for _, f := range fs {
+			if f != nil {
+				f(c)
+			}
+		}
+	}
+}
+
+func (c *conf) getNacosOptions() *NacosOptions {
+	if c.nacosOptions != nil {
+		return c.nacosOptions
+	}
+
+	if c.useNacos {
+		return &NacosOptions{
+			Address:   c.GetString(AddressKey, "127.0.0.1"),
+			Port:      c.GetInt(PortKey, 8848),
+			Namespace: c.GetString(Namespace, ""),
+			Group:     c.GetString(Group, ""),
+		}
+	}
+
+	return nil
+}

configure/configure_test.go

@@ -0,0 +1,42 @@
+package configure_test
+
+import (
+	"testing"
+
+	"github.com/charlienet/go-mixed/configure"
+	"github.com/charlienet/go-mixed/json"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestLoadSpecifiedFile(t *testing.T) {
+	conf, err := configure.New().SetConfigFile("config.toml").Read()
+	t.Log(err)
+
+	assert.Equal(t, "192.168.2.121", conf.GetString("nacos.address", ""))
+	_ = conf
+}
+
+func TestNewConfigure(t *testing.T) {
+
+}
+
+func TestNacos(t *testing.T) {
+	conf, err := configure.
+		New().
+		AddConfigPath(".").
+		WithNacos().
+		Read()
+
+	assert.Nil(t, err)
+
+	t.Log(conf.GetString("nacos.address", ""))
+
+	type redis struct {
+		Addrs string
+	}
+
+	r := &redis{}
+
+	t.Log(conf.Load("redis", r))
+	t.Log(json.StructToJsonIndent(r))
+}

configure/nacos.go

@@ -0,0 +1,96 @@
+package configure
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"github.com/nacos-group/nacos-sdk-go/v2/clients"
+	"github.com/nacos-group/nacos-sdk-go/v2/clients/config_client"
+	"github.com/nacos-group/nacos-sdk-go/v2/common/constant"
+	"github.com/nacos-group/nacos-sdk-go/v2/vo"
+)
+
+const (
+	AddressKey = "Nacos.Address"
+	PortKey    = "Nacos.Port"
+	Namespace  = "Nacos.Namespace"
+	Group      = "Nacos.Group"
+)
+
+type nacos struct {
+	client    config_client.IConfigClient
+	onChanged func(string, string)
+	group     string
+}
+
+type NacosOptions struct {
+	Address   string
+	Port      int
+	Namespace string
+	Group     string
+}
+
+func (c *conf) WithNacosOptions(options *NacosOptions) *conf {
+	c.nacosOptions = options
+	return c
+}
+
+func (c *conf) WithNacos() *conf {
+	c.useNacos = true
+	return c
+}
+
+func (n *nacos) Load(dataId string, v any) error {
+	voParam := vo.ConfigParam{
+		DataId:   dataId,
+		Group:    n.group,
+		OnChange: n.onChange,
+	}
+
+	content, err := n.client.GetConfig(voParam)
+	if err != nil {
+		return err
+	}
+
+	if len(content) == 0 {
+		return fmt.Errorf("parameters not configured:%s", dataId)
+	}
+
+	if err := json.Unmarshal([]byte(content), v); err != nil {
+		return err
+	}
+
+	n.client.ListenConfig(voParam)
+	return nil
+}
+
+func (n *nacos) onChange(namespace, group, dataId, data string) {
+	n.onChanged(dataId, data)
+}
+
+func createNacosClient(addr string, port int, namespace, group string) (config_client.IConfigClient, error) {
+	sc := []constant.ServerConfig{{
+		IpAddr: addr,
+		Port:   uint64(port),
+	}}
+
+	cc := constant.ClientConfig{
+		NamespaceId:         namespace,
+		TimeoutMs:           5000,
+		LogDir:              "logs",
+		CacheDir:            "cache",
+		LogLevel:            "info",
+		NotLoadCacheAtStart: true,
+	}
+
+	configClient, err := clients.CreateConfigClient(map[string]any{
+		"serverConfigs": sc,
+		"clientConfig":  cc,
+	})
+
+	if err != nil {
+		return nil, err
+	}
+
+	return configClient, nil
+}

locker/chan_source_locker.go

@@ -6,10 +6,17 @@ type ChanLocker interface {
 }
 
 type chanSourceLock struct {
-	m       RWLocker
+	m       rwLocker
 	content map[string]chan int
 }
 
+func NewChanSourceLocker() *chanSourceLock {
+	return &chanSourceLock{
+		m:       NewRWLocker(),
+		content: make(map[string]chan int),
+	}
+}
+
 func (s *chanSourceLock) Get(key string) (ch <-chan int, ok bool) {
 	s.m.RLock()
 	ch, ok = s.content[key]

locker/chan_source_locker_test.go

@@ -0,0 +1,19 @@
+package locker_test
+
+import (
+	"testing"
+
+	"github.com/charlienet/go-mixed/locker"
+)
+
+func TestChanSourceLocker(t *testing.T) {
+	l := locker.NewChanSourceLocker()
+	c, ok := l.Get("aaaa")
+	if ok {
+		<-c
+
+		println("ok")
+	}
+
+	println("fail")
+}

locker/distributed_locker.go

@@ -0,0 +1,7 @@
+package locker
+
+import "context"
+
+type DistributedLocker interface {
+	Unlock(context.Context, string)
+}

locker/distributed_locker_test.go

@@ -0,0 +1,13 @@
+package locker_test
+
+import (
+	"testing"
+
+	"github.com/charlienet/go-mixed/redis"
+	"github.com/charlienet/go-mixed/tests"
+)
+
+func TestRedisDistrbutedLocker(t *testing.T) {
+	tests.RunOnDefaultRedis(t, func(rdb redis.Client) {
+	})
+}

locker/empty_locker.go

@@ -1,14 +1,9 @@
 package locker
 
-var _ RWLocker = &emptyLocker{}
-var _ Locker = &emptyLocker{}
+var _ rwLocker = &emptyLocker{}
+var _ locker = &emptyLocker{}
 
-var EmptyLocker = &emptyLocker{}
-
-type emptyLocker struct{}
-
-func NewEmptyLocker() *emptyLocker {
-	return &emptyLocker{}
+type emptyLocker struct {
 }
 
 func (l *emptyLocker) RLock() {}

locker/locker.go

@@ -2,14 +2,16 @@ package locker
 
 import "sync"
 
-type Locker interface {
+type locker interface {
 	Lock()
 	Unlock()
 	TryLock() bool
 }
 
-type RWLocker interface {
-	Locker
+type rwLocker interface {
+	Lock()
+	Unlock()
+	TryLock() bool
 	RLock()
 	RUnlock()
 	TryRLock() bool
@@ -18,3 +20,7 @@ type RWLocker interface {
 func NewLocker() *sync.Mutex {
 	return &sync.Mutex{}
 }
+
+func NewRWLocker() *sync.RWMutex {
+	return &sync.RWMutex{}
+}

locker/readme.md

@@ -0,0 +1,22 @@
+同步锁
+
+EmptyLocker， 空锁
+RWLocker, 读写锁
+SpinLocker, 旋转锁
+
+
+锁可以添加一个外部存储成为分布式锁。WithRedis, WithZookeeper
+
+单例锁
+
+
+资源锁
+
+
+分布式锁
+在锁的基础上添加分布式存储升级为分布式锁 
+
+locker.WithRedis()
+locker.WithZookeeper()
+
+

locker/redis/redis_locker.lua

@@ -0,0 +1,33 @@
+#!lua name=charlie_locker
+
+-- 安装命令
+-- cat redis_locker.lua | redis-cli -x --cluster-only-masters --cluster call 192.168.123.30:6379 FUNCTION LOAD REPLACE
+
+local function lock(keys, args) 
+    if redis.call("GET", keys[1]) == args[1] then
+        redis.call("SET", keys[1], args[1], "PX", args[2])
+        return "OK"
+    else
+        return redis.call("SET", keys[1], args[1], "NX", "PX", args[2])
+    end
+end
+
+local function del(keys, args)
+    if redis.call("GET", keys[1]) == args[1] then
+        return redis.call("DEL", keys[1])
+    else
+        return '0'
+    end
+end
+
+local function expire(keys, args)
+    if redis.call('get', keys[1]) == args[1] then
+        return redis.call('expire', keys[1], args[2])
+    else
+        return '0'
+    end
+end
+
+redis.register_function('locker_lock',lock)
+redis.register_function('locker_unlock',del)
+redis.register_function('locker_expire',expire)

locker/redis/redis_store.go

@@ -0,0 +1,164 @@
+package redis
+
+import (
+	"context"
+	_ "embed"
+	"maps"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/charlienet/go-mixed/rand"
+	"github.com/charlienet/go-mixed/redis"
+	goredis "github.com/redis/go-redis/v9"
+)
+
+//go:embed redis_locker.lua
+var redis_locker_function string
+
+const (
+	defaultExpire = time.Second * 20
+	retryInterval = time.Millisecond * 10
+)
+
+var once sync.Once
+
+type redis_locker_store struct {
+	key     string
+	sources map[string]string
+	expire  time.Duration // 过期时间
+	mu      sync.RWMutex
+	clients []redis.Client
+}
+
+func NewRedisStore(key string, clients ...redis.Client) *redis_locker_store {
+	once.Do(func() { redis.Clients(clients).LoadFunction(redis_locker_function) })
+
+	locker := &redis_locker_store{
+		key:     key,
+		sources: make(map[string]string),
+		clients: clients,
+		expire:  defaultExpire,
+	}
+
+	go locker.expandLockTime()
+
+	return locker
+}
+
+func (l *redis_locker_store) Lock(ctx context.Context, sourceName string) error {
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+			if l.TryLock(ctx, sourceName) {
+				return nil
+			}
+		}
+
+		time.Sleep(retryInterval)
+	}
+}
+
+func (l *redis_locker_store) TryLock(ctx context.Context, sourceName string) bool {
+	value := l.getSourceValue(sourceName)
+
+	results := l.fCall(ctx, "locker_lock", sourceName, value, l.expire.Milliseconds())
+
+	if !isSuccess(results) {
+		for _, r := range results {
+			if r.Err() != nil {
+				println("err:", r.Err().Error())
+			}
+		}
+
+		l.Unlock(ctx, sourceName)
+		return false
+	}
+
+	return true
+}
+
+func (locker *redis_locker_store) Unlock(ctx context.Context, sourceName string) {
+	value := locker.getSourceValue(sourceName)
+	locker.fCall(ctx, "locker_unlock", sourceName, value)
+
+	locker.mu.Lock()
+	defer locker.mu.Unlock()
+
+	delete(locker.sources, sourceName)
+}
+
+func (l *redis_locker_store) expandLockTime() {
+	for {
+		time.Sleep(l.expire / 3)
+
+		if len(l.sources) == 0 {
+			continue
+		}
+
+		l.mu.RLock()
+		cloned := maps.Clone(l.sources)
+		l.mu.RUnlock()
+
+		for k, v := range cloned {
+			results := l.fCall(context.Background(), "locker_expire", k, v, l.expire.Seconds())
+			for _, r := range results {
+				if r.Err() != nil {
+					println("键延期失败:", r.Err().Error())
+				}
+			}
+		}
+	}
+}
+
+func (l *redis_locker_store) getSourceValue(name string) string {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+
+	if v, ok := l.sources[name]; ok {
+		return v
+	}
+
+	v := rand.Hex.Generate(36)
+	l.sources[name] = v
+	return v
+}
+
+func (locker *redis_locker_store) fCall(ctx context.Context, cmd string, key string, args ...any) []*goredis.Cmd {
+	results := make([]*goredis.Cmd, 0, len(locker.clients))
+
+	var wg sync.WaitGroup
+	wg.Add(len(locker.clients))
+	for _, rdb := range locker.clients {
+		go func(rdb redis.Client) {
+			defer wg.Done()
+
+			newKey := rdb.JoinKeys(locker.key, key)
+			results = append(results, rdb.FCall(ctx, cmd, []string{newKey}, args...))
+		}(rdb)
+	}
+
+	wg.Wait()
+
+	return results
+}
+
+func isSuccess(results []*goredis.Cmd) bool {
+	successCount := 0
+	for _, ret := range results {
+		resp, err := ret.Result()
+
+		if err != nil || resp == nil {
+			return false
+		}
+
+		reply, ok := resp.(string)
+		if ok && strings.EqualFold(reply, "OK") {
+			successCount++
+		}
+	}
+
+	return successCount >= len(results)/2+1
+}

locker/redis/redis_store_test.go

@@ -0,0 +1,31 @@
+package redis
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/charlienet/go-mixed/redis"
+	"github.com/charlienet/go-mixed/tests"
+)
+
+func TestCreateRedisStore(t *testing.T) {
+	tests.RunOnDefaultRedis(t, func(rdb redis.Client) {
+		keyName := "source"
+
+		l := NewRedisStore("locker_key", rdb)
+		ret := l.TryLock(context.Background(), keyName)
+		if !ret {
+			t.Log("加锁失败")
+		}
+
+		l.Lock(context.Background(), keyName)
+		t.Log("锁重入完成")
+
+		l.Unlock(context.Background(), keyName)
+
+		time.Sleep(time.Second * 15)
+
+		// l.Unlock(context.Background())
+	})
+}

locker/rw_locker.go

@@ -1,7 +0,0 @@
-package locker
-
-import "sync"
-
-func NewRWLocker() *sync.RWMutex {
-	return &sync.RWMutex{}
-}

locker/rw_locker_test.go

@@ -1,12 +0,0 @@
-package locker
-
-import "testing"
-
-func TestRWLokcer(t *testing.T) {
-	l := NewRWLocker()
-	l.RLock()
-
-	t.Log(l.TryRLock())
-
-	l.RUnlock()
-}

locker/source_locker.go

@@ -1,27 +1,45 @@
 package locker
 
 import (
+	"context"
 	"fmt"
 	"sync/atomic"
+
+	redis_store "github.com/charlienet/go-mixed/locker/redis"
+	"github.com/charlienet/go-mixed/redis"
 )
 
 // 带计数器锁
 type countLocker struct {
-	Locker
+	rw    rwLocker
 	Count int32
 }
 
 // SourceLocker 资源锁
 type SourceLocker struct {
-	m     RWLocker
-	locks map[string]*countLocker
+	m                 RWLocker
+	distributedLocker DistributedLocker
+	locks             map[string]*countLocker
+	err               error
 }
 
 func NewSourceLocker() *SourceLocker {
-	return &SourceLocker{
-		m:     NewRWLocker(),
+	l := &SourceLocker{
 		locks: make(map[string]*countLocker),
 	}
+
+	l.m.Synchronize()
+	return l
+}
+
+func (s *SourceLocker) WithRedis(key string, clients ...redis.Client) *SourceLocker {
+	redisStore := redis_store.NewRedisStore(key, clients...)
+	return s.WithDistributedLocker(redisStore)
+}
+
+func (s *SourceLocker) WithDistributedLocker(distributed DistributedLocker) *SourceLocker {
+	s.distributedLocker = distributed
+	return s
 }
 
 func (s *SourceLocker) Lock(key string) {
@@ -31,7 +49,7 @@ func (s *SourceLocker) Lock(key string) {
 
 	if ok {
 		atomic.AddInt32(&l.Count, 1)
-		l.Lock()
+		l.rw.Lock()
 
 		fmt.Println("加锁")
 	} else {
@@ -40,18 +58,15 @@ func (s *SourceLocker) Lock(key string) {
 		if l2, ok := s.locks[key]; ok {
 			s.m.Unlock()
 
-			l2.Lock()
+			l2.rw.Lock()
 			fmt.Println("二次检查加锁")
 		} else {
-			n := NewLocker()
-			s.locks[key] = &countLocker{Locker: n, Count: 1}
+			n := NewRWLocker()
+			s.locks[key] = &countLocker{rw: n, Count: 1}
 
 			s.m.Unlock()
 
-			fmt.Printf("新锁准备加锁:%p\n", n)
 			n.Lock()
-
-			fmt.Println("初始加锁")
 		}
 	}
 }
@@ -61,8 +76,11 @@ func (s *SourceLocker) Unlock(key string) {
 
 	if l, ok := s.locks[key]; ok {
 		atomic.AddInt32(&l.Count, -1)
-		fmt.Printf("解锁%p\n", l)
-		l.Unlock()
+		l.rw.Unlock()
+
+		if s.distributedLocker != nil {
+			s.distributedLocker.Unlock(context.Background(), key)
+		}
 
 		if l.Count == 0 {
 			delete(s.locks, key)
@@ -75,18 +93,14 @@ func (s *SourceLocker) TryLock(key string) bool {
 	// 加读锁
 	s.m.RLock()
 	l, ok := s.locks[key]
-
+	s.m.RUnlock()
 	if ok {
-		ret := l.TryLock()
-		s.m.RUnlock()
-
+		ret := l.rw.TryLock()
 		return ret
 	} else {
-		s.m.RUnlock()
-
 		s.m.Lock()
-		n := NewLocker()
-		s.locks[key] = &countLocker{Locker: n, Count: 1}
+		n := NewRWLocker()
+		s.locks[key] = &countLocker{rw: n, Count: 1}
 		s.m.Unlock()
 
 		return n.TryLock()

locker/source_locker_test.go

@@ -1,19 +1,39 @@
-package locker
+package locker_test
 
 import (
 	"sync"
 	"testing"
 	"time"
+
+	"github.com/charlienet/go-mixed/locker"
+	"github.com/stretchr/testify/assert"
 )
 
 var sourcekey = "u-0001"
 
 func TestTryLock(t *testing.T) {
+	l := locker.NewSourceLocker()
+	l.Lock("aa")
 
+	assert.False(t, l.TryLock("aa"))
+	assert.True(t, l.TryLock("bb"))
+
+	defer l.Unlock("aa")
+}
+
+func TestM(t *testing.T) {
+	l := locker.NewSourceLocker()
+
+	for i := 0; i < 10000000; i++ {
+		l.Lock("aaa")
+		l.Unlock("aaa")
+	}
+
+	t.Logf("%+v", l)
 }
 
 func TestSourceLocker(t *testing.T) {
-	l := NewSourceLocker()
+	l := locker.NewSourceLocker()
 
 	c := 5
 	n := 0
@@ -41,7 +61,7 @@ func TestSourceTryLock(t *testing.T) {
 	wg := new(sync.WaitGroup)
 	wg.Add(c)
 
-	l := NewSourceLocker()
+	l := locker.NewSourceLocker()
 
 	for i := 0; i < c; i++ {
 		go func() {
@@ -61,7 +81,7 @@ func TestSourceTryLock(t *testing.T) {
 }
 
 func BenchmarkSourceLocker(b *testing.B) {
-	l := NewSourceLocker()
+	l := locker.NewSourceLocker()
 
 	b.RunParallel(func(p *testing.PB) {
 		for p.Next() {

locker/spin_locker_test.go

@@ -1,12 +1,14 @@
-package locker
+package locker_test
 
 import (
 	"sync"
 	"testing"
+
+	"github.com/charlienet/go-mixed/locker"
 )
 
 func TestSpinLock(t *testing.T) {
-	l := NewSpinLocker()
+	l := locker.NewSpinLocker()
 
 	n := 10
 	c := 0

locker/synchronizeable.go

@@ -3,94 +3,111 @@ package locker
 import (
 	"log"
 	"sync"
+
+	"github.com/charlienet/go-mixed/redis"
 )
 
-type WithLocker struct {
-	once sync.Once
-	mu   Locker
+var empty = &emptyLocker{}
+
+type Locker struct {
+	once              sync.Once
+	distributedLocker DistributedLocker // 分布式锁
+	mu                locker
 }
 
-func (w *WithLocker) Synchronize() {
-	if w.mu == nil || w.mu == EmptyLocker {
+func (w *Locker) WithRedis(key string, rdb redis.Client) *Locker {
+	return w
+}
+
+func (w *Locker) WithDistributedLocker(d DistributedLocker) *Locker {
+	return w
+}
+
+func (w *Locker) Synchronize() *Locker {
+	if w.mu == nil || w.mu == empty {
 		w.mu = NewLocker()
 	}
+
+	return w
 }
 
-func (w *WithLocker) Lock() {
-	w.ensureLocker().Lock()
+func (w *Locker) Lock() {
+	w.ensureLocker().mu.Lock()
 }
 
-func (w *WithLocker) Unlock() {
-	w.ensureLocker().Unlock()
+func (w *Locker) Unlock() {
+	w.ensureLocker().mu.Unlock()
 }
 
-func (w *WithLocker) TryLock() bool {
-	return w.ensureLocker().TryLock()
+func (w *Locker) TryLock() bool {
+	return w.ensureLocker().mu.TryLock()
 }
 
-func (w *WithLocker) ensureLocker() Locker {
+func (w *Locker) ensureLocker() *Locker {
 	w.once.Do(func() {
 		if w.mu == nil {
-			w.mu = EmptyLocker
+			w.mu = empty
 		}
+
 	})
 
-	return w.mu
+	return w
 }
 
-type WithSpinLocker struct {
-	WithLocker
+type SpinLocker struct {
+	Locker
 }
 
-func (w *WithSpinLocker) Synchronize() {
-	if w.mu == nil || w.mu == EmptyLocker {
+func (w *SpinLocker) Synchronize() {
+	if w.mu == nil || w.mu == empty {
 		w.mu = NewSpinLocker()
 	}
 }
 
-type WithRWLocker struct {
+type RWLocker struct {
 	once sync.Once
-	mu   RWLocker
+	mu   rwLocker
 }
 
-func (w *WithRWLocker) Synchronize() {
-	if w.mu == nil || w.mu == EmptyLocker {
-		log.Println("初始化有效锁")
+func (w *RWLocker) Synchronize() *RWLocker {
+	if w.mu == nil || w.mu == empty {
 		w.mu = NewRWLocker()
 	}
+
+	return w
 }
 
-func (w *WithRWLocker) Lock() {
-	w.ensureLocker().Lock()
+func (w *RWLocker) Lock() {
+	w.ensureLocker().mu.Lock()
 }
 
-func (w *WithRWLocker) TryLock() bool {
-	return w.ensureLocker().TryLock()
+func (w *RWLocker) TryLock() bool {
+	return w.ensureLocker().mu.TryLock()
 }
 
-func (w *WithRWLocker) Unlock() {
-	w.ensureLocker().Unlock()
+func (w *RWLocker) Unlock() {
+	w.ensureLocker().mu.Unlock()
 }
 
-func (w *WithRWLocker) RLock() {
-	w.ensureLocker().RLock()
+func (w *RWLocker) RLock() {
+	w.ensureLocker().mu.RLock()
 }
 
-func (w *WithRWLocker) TryRLock() bool {
-	return w.ensureLocker().TryRLock()
+func (w *RWLocker) TryRLock() bool {
+	return w.ensureLocker().mu.TryRLock()
 }
 
-func (w *WithRWLocker) RUnlock() {
-	w.ensureLocker().RUnlock()
+func (w *RWLocker) RUnlock() {
+	w.ensureLocker().mu.RUnlock()
 }
 
-func (w *WithRWLocker) ensureLocker() RWLocker {
+func (w *RWLocker) ensureLocker() *RWLocker {
 	w.once.Do(func() {
 		if w.mu == nil {
 			log.Println("初始化一个空锁")
-			w.mu = EmptyLocker
+			w.mu = empty
 		}
 	})
 
-	return w.mu
+	return w
 }

locker/synchronizeable_test.go

@@ -0,0 +1,44 @@
+package locker_test
+
+import (
+	"testing"
+
+	"github.com/charlienet/go-mixed/locker"
+)
+
+func TestLocker(t *testing.T) {
+
+	var l locker.Locker
+
+	l.Synchronize()
+
+	l.Lock()
+	defer l.Unlock()
+}
+
+func TestNew(t *testing.T) {
+	var a locker.RWLocker
+	a.Synchronize()
+
+}
+
+func TestSpinLocker(t *testing.T) {
+	var l locker.SpinLocker
+	l.Synchronize()
+
+	l.Lock()
+	defer l.Unlock()
+}
+
+func TestRWLocker(t *testing.T) {
+	var l locker.RWLocker
+	l.Lock()
+}
+
+func TestPointLocker(t *testing.T) {
+	l := locker.NewLocker()
+	l.Lock()
+	l.Lock()
+
+	defer l.Unlock()
+}

sets/hash_set.go

@@ -13,14 +13,13 @@ import (
 var _ Set[string] = &hash_set[string]{}
 
 type hash_set[T constraints.Ordered] struct {
-	m    map[T]struct{}
-	lock locker.RWLocker
+	m      map[T]struct{}
+	locker locker.RWLocker
 }
 
 func NewHashSet[T constraints.Ordered](values ...T) *hash_set[T] {
 	set := hash_set[T]{
-		m:    make(map[T]struct{}, len(values)),
-		lock: locker.EmptyLocker,
+		m: make(map[T]struct{}, len(values)),
 	}
 
 	set.Add(values...)
@@ -28,37 +27,41 @@ func NewHashSet[T constraints.Ordered](values ...T) *hash_set[T] {
 }
 
 func (s *hash_set[T]) Sync() *hash_set[T] {
-	s.lock = locker.NewRWLocker()
+	s.locker.Synchronize()
 	return s
 }
 
-func (s hash_set[T]) Add(values ...T) {
-	s.lock.Lock()
-	defer s.lock.Unlock()
+func (s *hash_set[T]) Add(values ...T) Set[T] {
+	s.locker.Lock()
+	defer s.locker.Unlock()
 
 	for _, v := range values {
 		s.m[v] = struct{}{}
 	}
+
+	return s
 }
 
-func (s hash_set[T]) Remove(v T) {
-	s.lock.Lock()
-	defer s.lock.Unlock()
+func (s *hash_set[T]) Remove(v T) Set[T] {
+	s.locker.Lock()
+	defer s.locker.Unlock()
 
 	delete(s.m, v)
+
+	return s
 }
 
-func (s hash_set[T]) Contains(value T) bool {
-	s.lock.RLock()
-	defer s.lock.RUnlock()
+func (s *hash_set[T]) Contains(value T) bool {
+	s.locker.RLock()
+	defer s.locker.RUnlock()
 
 	_, ok := s.m[value]
 	return ok
 }
 
-func (s hash_set[T]) ContainsAny(values ...T) bool {
-	s.lock.RLock()
-	defer s.lock.RUnlock()
+func (s *hash_set[T]) ContainsAny(values ...T) bool {
+	s.locker.RLock()
+	defer s.locker.RUnlock()
 
 	for _, v := range values {
 		if _, ok := s.m[v]; ok {
@@ -69,9 +72,9 @@ func (s hash_set[T]) ContainsAny(values ...T) bool {
 	return false
 }
 
-func (s hash_set[T]) ContainsAll(values ...T) bool {
-	s.lock.RLock()
-	defer s.lock.RUnlock()
+func (s *hash_set[T]) ContainsAll(values ...T) bool {
+	s.locker.RLock()
+	defer s.locker.RUnlock()
 
 	for _, v := range values {
 		if _, ok := s.m[v]; !ok {
@@ -82,15 +85,15 @@ func (s hash_set[T]) ContainsAll(values ...T) bool {
 	return true
 }
 
-func (s hash_set[T]) Asc() Set[T] {
+func (s *hash_set[T]) Asc() Set[T] {
 	return s.copyToSorted().Asc()
 }
 
-func (s hash_set[T]) Desc() Set[T] {
+func (s *hash_set[T]) Desc() Set[T] {
 	return s.copyToSorted().Desc()
 }
 
-func (s hash_set[T]) copyToSorted() Set[T] {
+func (s *hash_set[T]) copyToSorted() Set[T] {
 	orderd := NewSortedSet[T]()
 	for k := range s.m {
 		orderd.Add(k)
@@ -109,13 +112,13 @@ func (s *hash_set[T]) Clone() *hash_set[T] {
 	return set
 }
 
-func (s hash_set[T]) Iterate(fn func(value T)) {
+func (s *hash_set[T]) Iterate(fn func(value T)) {
 	for v := range s.m {
 		fn(v)
 	}
 }
 
-func (s hash_set[T]) ToSlice() []T {
+func (s *hash_set[T]) ToSlice() []T {
 	values := make([]T, 0, s.Size())
 	s.Iterate(func(value T) {
 		values = append(values, value)
@@ -124,15 +127,15 @@ func (s hash_set[T]) ToSlice() []T {
 	return values
 }
 
-func (s hash_set[T]) IsEmpty() bool {
+func (s *hash_set[T]) IsEmpty() bool {
 	return len(s.m) == 0
 }
 
-func (s hash_set[T]) Size() int {
+func (s *hash_set[T]) Size() int {
 	return len(s.m)
 }
 
-func (s hash_set[T]) MarshalJSON() ([]byte, error) {
+func (s *hash_set[T]) MarshalJSON() ([]byte, error) {
 	items := make([]string, 0, s.Size())
 
 	for ele := range s.m {
@@ -147,7 +150,7 @@ func (s hash_set[T]) MarshalJSON() ([]byte, error) {
 	return []byte(fmt.Sprintf("[%s]", strings.Join(items, ", "))), nil
 }
 
-func (s hash_set[T]) UnmarshalJSON(b []byte) error {
+func (s *hash_set[T]) UnmarshalJSON(b []byte) error {
 	var i []any
 
 	d := json.NewDecoder(bytes.NewReader(b))
@@ -166,7 +169,7 @@ func (s hash_set[T]) UnmarshalJSON(b []byte) error {
 	return nil
 }
 
-func (s hash_set[T]) String() string {
+func (s *hash_set[T]) String() string {
 	l := make([]string, 0, len(s.m))
 	for k := range s.m {
 		l = append(l, fmt.Sprint(k))

sets/set.go

@@ -1,15 +1,13 @@
 package sets
 
 import (
-	"sync"
-
 	"github.com/charlienet/go-mixed/locker"
 	"golang.org/x/exp/constraints"
 )
 
 type Set[T comparable] interface {
-	Add(...T)
-	Remove(v T)
+	Add(...T) Set[T]
+	Remove(v T) Set[T]
 	Asc() Set[T]
 	Desc() Set[T]
 	Contains(T) bool
@@ -19,18 +17,16 @@ type Set[T comparable] interface {
 	ToSlice() []T // 转换为切片
 }
 
-var defaultOptions = option{locker: locker.NewEmptyLocker()}
-
 type option struct {
-	locker sync.Locker
+	locker locker.Locker
 }
 
-type setFunc func(option)
+type setFunc func(*option)
 
 func WithSync() setFunc {
 
-	return func(o option) {
-		o.locker = &sync.RWMutex{}
+	return func(o *option) {
+		o.locker.Synchronize()
 	}
 }
 

sets/sorted_set.go

@@ -14,20 +14,23 @@ type sorted_set[T constraints.Ordered] struct {
 
 func NewSortedSet[T constraints.Ordered](t ...T) *sorted_set[T] {
 	return &sorted_set[T]{
-		set: NewHashSet[T](),
+		sorted: t,
+		set:    NewHashSet(t...),
 	}
 }
 
-func (s *sorted_set[T]) Add(values ...T) {
+func (s *sorted_set[T]) Add(values ...T) Set[T] {
 	for _, v := range values {
 		if !s.set.Contains(v) {
 			s.sorted = append(s.sorted, v)
 			s.set.Add(v)
 		}
 	}
+
+	return s
 }
 
-func (s *sorted_set[T]) Remove(v T) {
+func (s *sorted_set[T]) Remove(v T) Set[T] {
 	if s.set.Contains(v) {
 		for index := range s.sorted {
 			if s.sorted[index] == v {
@@ -38,6 +41,8 @@ func (s *sorted_set[T]) Remove(v T) {
 
 		s.set.Remove(v)
 	}
+
+	return s
 }
 
 func (s *sorted_set[T]) Asc() Set[T] {