支持缓存+队列，支持驱动：redis、memory

pkg/cache/doc.go

@@ -0,0 +1,6 @@
+package cache
+
+// 缓存+队列套件，redis队列使用时，请参考单元测试中的配置
+// 目前支持:
+// - redis
+// - memory

pkg/cache/memory.go

@@ -0,0 +1,15 @@
+package cache
+
+import (
+	"github.com/matchstalk/utils/cache"
+)
+
+var MemoryAdapter Adapter
+
+func InitMemory() error {
+	MemoryAdapter = &cache.Memory{
+		PoolNum: 100,
+	}
+	err := MemoryAdapter.Connect()
+	return err
+}

pkg/cache/memory_test.go

@@ -0,0 +1,45 @@
+package cache
+
+import (
+	"fmt"
+	"github.com/matchstalk/utils/cache"
+	"testing"
+	"time"
+)
+
+func TestInitMemory(t *testing.T) {
+	tests := []struct {
+		name    string
+		wantErr bool
+	}{
+		{
+			"test01",
+			false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if err := InitMemory(); (err != nil) != tt.wantErr {
+				t.Errorf("InitRedis() error = %v, wantErr %v", err, tt.wantErr)
+			}
+			MemoryAdapter.Set("test", "1", 100)
+			key, _ := MemoryAdapter.Get("test")
+			message := &cache.MemoryMessage{}
+			message.Stream = "queuetest"
+			message.Values = map[string]interface{}{
+				"key": "value",
+			}
+			MemoryAdapter.Append("queuetest", message)
+			MemoryAdapter.Register("queuetest", func(message cache.Message) error {
+				fmt.Println(message.GetValues())
+				return nil
+			})
+			go func() {
+				MemoryAdapter.Run()
+			}()
+			time.Sleep(time.Second)
+			MemoryAdapter.Shutdown()
+			fmt.Println(key)
+		})
+	}
+}

pkg/cache/redis.go

@@ -0,0 +1,31 @@
+package cache
+
+import (
+	"github.com/go-redis/redis/v7"
+	"github.com/matchstalk/redisqueue"
+	"github.com/matchstalk/utils/cache"
+	"time"
+)
+
+var RedisAdapter Adapter
+
+func InitRedis() error {
+	RedisAdapter = &cache.Redis{
+		ConnectOption: &redis.Options{
+			Addr: "127.0.0.1:6379",
+		},
+		ConsumerOptions: &redisqueue.ConsumerOptions{
+			VisibilityTimeout: 60 * time.Second,
+			BlockingTimeout:   5 * time.Second,
+			ReclaimInterval:   1 * time.Second,
+			BufferSize:        100,
+			Concurrency:       10,
+		},
+		ProducerOptions: &redisqueue.ProducerOptions{
+			StreamMaxLength:      100,
+			ApproximateMaxLength: true,
+		},
+	}
+	err := RedisAdapter.Connect()
+	return err
+}

pkg/cache/redis_test.go

@@ -0,0 +1,45 @@
+package cache
+
+import (
+	"fmt"
+	"github.com/matchstalk/utils/cache"
+	"testing"
+	"time"
+)
+
+func TestInitRedis(t *testing.T) {
+	tests := []struct {
+		name    string
+		wantErr bool
+	}{
+		{
+			"test01",
+			false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if err := InitRedis(); (err != nil) != tt.wantErr {
+				t.Errorf("InitRedis() error = %v, wantErr %v", err, tt.wantErr)
+			}
+			RedisAdapter.Set("test", "1", 100)
+			key, _ := RedisAdapter.Get("test")
+			message := &cache.RedisMessage{}
+			message.Stream = "queuetest"
+			message.Values = map[string]interface{}{
+				"key": "value",
+			}
+			RedisAdapter.Append("queuetest", message)
+			RedisAdapter.Register("queuetest", func(message cache.Message) error {
+				fmt.Println(message.GetValues())
+				return nil
+			})
+			go func() {
+				RedisAdapter.Run()
+			}()
+			time.Sleep(time.Second)
+			RedisAdapter.Shutdown()
+			fmt.Println(key)
+		})
+	}
+}

pkg/cache/type.go

@@ -2,6 +2,8 @@ package cache
 
 import (
 	"time"
+
+	"github.com/matchstalk/utils/cache"
 )
 
 type Adapter interface {
@@ -12,9 +14,9 @@ type Adapter interface {
 	HashGet(hk, key string) (string, error)
 	HashDel(hk, key string) error
 	Increase(key string) error
+	Decrease(key string) error
 	Expire(key string, dur time.Duration) error
-	SetQueue(name string, message Message) error
-	GetQueue(name string, f func(message Message) error)
+	cache.AdapterQueue
 }
 
 type Message interface {