Redis 高级特性

127.0.0.1:6379> MULTI
OK
127.0.0.1:6379> SET user:1:name "张三"
QUEUED
127.0.0.1:6379> SET user:1:age "28"
QUEUED
127.0.0.1:6379> INCR user:1:visits
QUEUED
127.0.0.1:6379> EXEC
1) OK
2) OK
3) (integer) 1

127.0.0.1:6379> MULTI
OK
127.0.0.1:6379> SET temp "will be discarded"
QUEUED
127.0.0.1:6379> DISCARD
OK
127.0.0.1:6379> GET temp
(nil)

# 终端 1：监视并开启事务
127.0.0.1:6379> WATCH stock:1001
OK
127.0.0.1:6379> GET stock:1001
"100"
127.0.0.1:6379> MULTI
OK
127.0.0.1:6379> DECR stock:1001
QUEUED
# 此时终端 2 执行了: SET stock:1001 99
127.0.0.1:6379> EXEC
(nil)   # 事务被取消，因为 stock:1001 已被修改

# 终端 2：在终端 1 的事务提交前修改
127.0.0.1:6379> SET stock:1001 99
OK

127.0.0.1:6379> MULTI
OK
127.0.0.1:6379> SET key1 "value1"
QUEUED
127.0.0.1:6379> INVALID_COMMAND   # 不存在的命令
(error) ERR unknown command 'INVALID_COMMAND'
127.0.0.1:6379> SET key2 "value2"
QUEUED
127.0.0.1:6379> EXEC
(error) EXECABORT Transaction discarded because of previous errors.
127.0.0.1:6379> GET key1
(nil)   # 第一条也没执行，全部被丢弃

127.0.0.1:6379> SET str_key "hello"
OK
127.0.0.1:6379> MULTI
OK
127.0.0.1:6379> LPUSH str_key "world"   # 对 String 执行 List 操作
QUEUED
127.0.0.1:6379> SET key2 "good"
QUEUED
127.0.0.1:6379> EXEC
1) (error) WRONGTYPE Operation against a key holding the wrong kind of value
2) OK                                  # key2 仍然被正确设置
127.0.0.1:6379> GET key2
"good"

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/redis/go-redis/v9"
)

func deductStock(ctx context.Context, rdb *redis.Client, key string) error {
	txf := func(tx *redis.Tx) error {
		// 在事务外获取当前值
		n, err := tx.Get(ctx, key).Int()
		if err != nil && err != redis.Nil {
			return err
		}
		if n <= 0 {
			return fmt.Errorf("库存不足")
		}

		// 事务内操作：WATCH 之后到 EXEC 之间如果 key 被改，则返回 error
		_, err = tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
			pipe.Decr(ctx, key)
			return nil
		})
		return err
	}

	// 重试机制：WATCH 失败时自动重试
	for i := 0; i < 3; i++ {
		err := rdb.Watch(ctx, txf, key)
		if err == nil {
			return nil // 成功
		}
		if err == redis.TxFailedErr {
			// WATCH 失败（事务冲突），重试
			continue
		}
		return err // 其他错误直接返回
	}
	return fmt.Errorf("多次重试后仍失败")
}

func main() {
	ctx := context.Background()
	rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379"})

	rdb.Set(ctx, "stock:1001", 100, 0)

	var wg sync.WaitGroup
	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			err := deductStock(ctx, rdb, "stock:1001")
			if err != nil {
				fmt.Println("扣库存失败:", err)
			} else {
				fmt.Println("扣库存成功")
			}
		}()
	}
	wg.Wait()
}

EVAL script numkeys key [key ...] arg [arg ...]

127.0.0.1:6379> EVAL "return redis.call('SET', KEYS[1], ARGV[1])" 1 mykey myvalue
OK
127.0.0.1:6379> GET mykey
"myvalue"

EVALSHA sha1 numkeys key [key ...] arg [arg ...]

127.0.0.1:6379> SCRIPT LOAD "return redis.call('GET', KEYS[1])"
"a5260dd66c02f97f8c11a3a42ab3e5f7d45a5a12"
127.0.0.1:6379> EVALSHA a5260dd66c02f97f8c11a3a42ab3e5f7d45a5a12 1 mykey
"myvalue"

127.0.0.1:6379> EVALSHA deadbeef 1 mykey
(error) NOSCRIPT No matching script. Please use EVAL.

-- redis.call()：命令失败时抛出 Lua 错误，脚本终止
local v = redis.call('GET', KEYS[1])

-- redis.pcall()：命令失败时返回 error 对象，脚本可继续
local result = redis.pcall('GET', KEYS[1])
if result.err then
    return "key not found"
end

-- KEYS[1], KEYS[2] ... 不是 KEYS[0]
local val = redis.call('GET', KEYS[1])
-- ARGV[1], ARGV[2] ...
local limit = tonumber(ARGV[1])

127.0.0.1:6379> SCRIPT KILL
(error) UNKILLABLE Sorry the script already executed write commands against the dataset. You can either wait the script termination or kill the server in a hard way using the SHUTDOWN NOSAVE command.

-- KEYS[1]: 锁的 key
-- ARGV[1]: 锁的 value（持有者标识，如 UUID）
if redis.call('get', KEYS[1]) == ARGV[1] then
    return redis.call('del', KEYS[1])
else
    return 0
end

# EVAL script 1 lock:order:1001 "my-uuid-1234"
127.0.0.1:6379> SET lock:order:1001 "my-uuid-1234" NX EX 30
OK
127.0.0.1:6379> EVAL "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end" 1 lock:order:1001 "my-uuid-1234"
(integer) 1

-- KEYS[1]: 限流 key (如 rate:user:1001)
-- ARGV[1]: 窗口内允许的最大请求数
-- ARGV[2]: 窗口大小（秒）
local key = KEYS[1]
local limit = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local now = tonumber(redis.call('TIME')[1])     -- 当前 Unix 秒

-- 窗口起始时间
local window_start = now - window

-- 移除窗口外的旧记录
redis.call('ZREMRANGEBYSCORE', key, 0, window_start)

-- 当前窗口内的请求数
local current = redis.call('ZCARD', key)

if current < limit then
    redis.call('ZADD', key, now, now .. ':' .. ARGV[3])  -- ARGV[3] 为请求 ID 去重
    redis.call('EXPIRE', key, window + 1)
    return 1   -- 放行
else
    return 0   -- 限流
end

EVAL "..." 1 rate:user:1001 5 60 req-001

-- KEYS[1]: 库存 key (如 seckill:stock:1001)
-- KEYS[2]: 购买记录 key (如 seckill:users:1001)
-- ARGV[1]: 用户 ID
local userId = ARGV[1]
local stockKey = KEYS[1]
local orderKey = KEYS[2]

-- 1. 检查库存
local stock = tonumber(redis.call('get', stockKey) or 0)
if stock <= 0 then
    return 0   -- 库存不足
end

-- 2. 检查是否已购买（一人一单）
local exist = redis.call('sismember', orderKey, userId)
if exist == 1 then
    return 2   -- 已购买过了
end

-- 3. 扣库存 + 标记已购买
redis.call('decr', stockKey)
redis.call('sadd', orderKey, userId)
return 1       -- 抢购成功

const luaSeckill = `
local userId = ARGV[1]
local stockKey = KEYS[1]
local orderKey = KEYS[2]
local stock = tonumber(redis.call('get', stockKey) or 0)
if stock <= 0 then return 0 end
if redis.call('sismember', orderKey, userId) == 1 then return 2 end
redis.call('decr', stockKey)
redis.call('sadd', orderKey, userId)
return 1
`

func seckill(ctx context.Context, rdb *redis.Client, stockKey, orderKey, userId string) (int, error) {
    result, err := rdb.Eval(ctx, luaSeckill, []string{stockKey, orderKey}, userId).Int()
    if err != nil {
        return -1, err
    }
    switch result {
    case 0: fmt.Println("库存不足")
    case 1: fmt.Println("抢购成功")
    case 2: fmt.Println("已经购买过了")
    }
    return result, nil
}

func EvalWithFallback(ctx context.Context, rdb *redis.Client, sha, script string, keys []string, args ...interface{}) *redis.Cmd {
    // 优先使用 EVALSHA（节省带宽）
    result := rdb.EvalSha(ctx, sha, keys, args...)
    if err := result.Err(); err != nil && strings.HasPrefix(err.Error(), "NOSCRIPT") {
        // 缓存未命中，回退到 EVAL
        result = rdb.Eval(ctx, script, keys, args...)
    }
    return result
}

# 加载函数库
127.0.0.1:6379> FUNCTION LOAD "#!lua name=mylib\nredis.register_function('my_get', function(keys, args) return redis.call('GET', keys[1]) end)"

# 列出所有函数
127.0.0.1:6379> FUNCTION LIST
1) 1) "library_name"
   2) "mylib"
   3) "engine"
   4) "LUA"
   5) "functions"
   6) 1) 1) "name"
         2) "my_get"
         3) "description"
         4) (nil)
         5) "flags"
         6) (empty array)

# 调用函数（FCALL 代替 EVAL）
127.0.0.1:6379> FCALL my_get 1 mykey
"myvalue"

# 删除函数库
127.0.0.1:6379> FUNCTION DELETE mylib
OK

-- EVAL 方式的函数签名
-- EVAL "..." 2 key1 key2 arg1 arg2
-- 脚本中: KEYS[1]=key1, KEYS[2]=key2, ARGV[1]=arg1, ARGV[2]=arg2

-- FUNCTION LOAD 方式的函数签名
-- redis.register_function('myfunc', function(keys, args)
--     local key1 = keys[1]
--     local arg1 = args[1]
-- end)
-- FCALL myfunc 2 key1 key2 arg1 arg2

#!lua name=cache_lib

-- 缓存读取 + 自动回源（简化版 Cache-Aside）
redis.register_function(
    'cache_get_or_set',
    function(keys, args)
        local cache_key = keys[1]
        local ttl = tonumber(args[1])

        local cached = redis.call('GET', cache_key)
        if cached then
            return cached
        end

        -- 回源逻辑由调用方通过 args 传入标志位控制
        -- 这里简化为返回 nil，由客户端回源
        return false
    end
)

package main

import (
	"context"
	"fmt"

	"github.com/redis/go-redis/v9"
)

func main() {
	ctx := context.Background()
	rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379"})

	// 方式一：Pipeline() - 手动管理
	pipe := rdb.Pipeline()

	// 入队命令（尚未发送到 Redis）
	set1 := pipe.Set(ctx, "user:1:name", "Alice", 0)
	set2 := pipe.Set(ctx, "user:1:age", 30, 0)
	incr := pipe.Incr(ctx, "user:1:logins")

	// Exec() 一次性发送所有命令并接收结果
	cmds, err := pipe.Exec(ctx)
	if err != nil {
		panic(err)
	}

	// 逐个获取结果
	for i, cmd := range cmds {
		fmt.Printf("cmd[%d]: %v\n", i, cmd)
	}
	fmt.Println("set1:", set1.Val())  // OK
	fmt.Println("set2:", set2.Val())  // OK
	fmt.Println("incr:", incr.Val())  // 1
}

// 方式二：Pipelined() - 回调式，自动 Exec
var incr *redis.IntCmd
_, err := rdb.Pipelined(ctx, func(pipe redis.Pipeliner) error {
	pipe.Set(ctx, "key1", "val1", 0)
	pipe.Set(ctx, "key2", "val2", 0)
	incr = pipe.Incr(ctx, "counter")
	return nil
})
if err != nil {
	panic(err)
}
fmt.Println("counter:", incr.Val())

// 批量获取多个 key 的值
func batchGet(ctx context.Context, rdb *redis.Client, keys []string) (map[string]string, error) {
    pipe := rdb.Pipeline()
    cmds := make(map[string]*redis.StringCmd, len(keys))

    for _, key := range keys {
        cmds[key] = pipe.Get(ctx, key)
    }

    _, err := pipe.Exec(ctx)
    if err != nil && err != redis.Nil {
        return nil, err
    }

    result := make(map[string]string, len(keys))
    for key, cmd := range cmds {
        val, err := cmd.Result()
        if err == redis.Nil {
            continue // key 不存在跳过
        }
        if err != nil {
            return nil, err
        }
        result[key] = val
    }
    return result, nil
}

# 启动时加载
redis-server --loadmodule /path/to/module.so

# 运行时加载
redis-cli MODULE LOAD /path/to/module.so

# 查看已加载模块
127.0.0.1:6379> MODULE LIST
1) 1) "name"
   2) "bf"
   3) "ver"
   4) (integer) 20600

# 创建布隆过滤器（容量 1000，误判率 0.01）
BF.RESERVE users:bloom 0.01 1000

# 添加元素
BF.ADD users:bloom "user:1001"
BF.MADD users:bloom "user:1002" "user:1003" "user:1004"

# 检查元素是否存在
BF.EXISTS users:bloom "user:1001"
# (integer) 1    -- 可能存在（不会误判为不存在）

BF.EXISTS users:bloom "user:9999"
# (integer) 0    -- 一定不存在

CF.RESERVE items:cf 1000
CF.ADD items:cf "item-001"
CF.EXISTS items:cf "item-001"
CF.DEL items:cf "item-001"
CF.COUNT items:cf "item-001"

CMS.INITBYDIM top:pages 2000 5
CMS.INCRBY top:pages "/home" 1
CMS.INCRBY top:pages "/products" 5
CMS.QUERY top:pages "/home"

TOPK.RESERVE top:tags 10 2000 7 0.925
TOPK.ADD top:tags "golang" "redis" "python"
TOPK.LIST top:tags

# 设置 JSON 文档
JSON.SET user:1001 $ '{"name":"张三","age":28,"tags":["golang","redis"]}'

# 获取整个文档
JSON.GET user:1001 $
# [{"name":"张三","age":28,"tags":["golang","redis"]}]

# 获取指定路径（JSONPath）
JSON.GET user:1001 $.name
# ["张三"]

JSON.GET user:1001 $.tags[0]
# ["golang"]

# 修改指定字段
JSON.SET user:1001 $.age 29

# 数值递增
JSON.NUMINCRBY user:1001 $.age 1
# [30]

# 数组操作
JSON.ARRAPPEND user:1001 $.tags '"kafka"'
JSON.ARRLEN user:1001 $.tags
JSON.ARRINDEX user:1001 $.tags '"redis"'

# 字符串追加
JSON.STRAPPEND user:1001 $.name ' (工程师)'

# 删除
JSON.DEL user:1001 $.tags[-1]

# 创建索引
FT.CREATE idx:articles ON HASH PREFIX 1 article: SCHEMA title TEXT WEIGHT 5.0 body TEXT author TAG

# 添加文档
HSET article:1 title "Redis入门指南" body "Redis是一个开源的内存数据库..." author "张三"
HSET article:2 title "Redis高级特性" body "本文介绍Redis事务与Lua脚本..." author "李四"

# 全文搜索
FT.SEARCH idx:articles "Redis"
FT.SEARCH idx:articles "@title:Redis"
FT.SEARCH idx:articles "@author:{张三}"

# 聚合查询
FT.AGGREGATE idx:articles "*" GROUPBY 1 @author REDUCE COUNT 0 AS count

# 创建时序
TS.CREATE temp:sensor:1 RETENTION 86400000 LABELS sensor_id 1 type temperature

# 添加数据点
TS.ADD temp:sensor:1 * 23.5
TS.ADD temp:sensor:1 * 24.1

# 按范围查询
TS.RANGE temp:sensor:1 1650000000 1650086400

# 聚合查询（按小时平均）
TS.RANGE temp:sensor:1 1650000000 1650086400 AGGREGATION avg 3600000

# 降采样规则（写入时自动聚合）
TS.CREATE temp:sensor:1:avg
TS.CREATERULE temp:sensor:1 temp:sensor:1:avg AGGREGATION avg 60000

# 多 key 查询
TS.MRANGE 1650000000 1650086400 AGGREGATION avg 3600000 FILTER type=temperature

# 创建节点和关系
GRAPH.QUERY social "CREATE (:Person {name:'Alice', age:30})"
GRAPH.QUERY social "CREATE (:Person {name:'Bob', age:32})"
GRAPH.QUERY social "MATCH (a:Person {name:'Alice'}), (b:Person {name:'Bob'}) CREATE (a)-[:FRIEND]->(b)"

# 查询（Cypher 语法）
GRAPH.QUERY social "MATCH (a:Person)-[:FRIEND]->(b:Person) RETURN a.name, b.name"
GRAPH.QUERY social "MATCH (a:Person) WHERE a.age > 30 RETURN a.name, a.age"

# RedisGears 使用 Python 编写处理逻辑
def process(x):
    redis.call('INCR', 'processed_count')
    return x

GB('StreamReader').map(process).register()

# 列出所有 ACL 规则
ACL LIST
# 1) "user default on nopass ~* &* +@all"
# 2) "user app on >password123 ~cache:* +@all -@dangerous"

# 列出所有用户
ACL USERS
# 1) "default"
# 2) "app"

# 查看当前用户
ACL WHOAMI
# "default"

# 查看用户详情
ACL GETUSER app
# 1) "flags"
# 2) (empty array)
# 3) "passwords"
# 4) 1) "..."
# ...

ACL SETUSER <username> [规则...]

# 创建只读用户
ACL SETUSER readonly_user on >readonly_pass ~* +@read -@write -@dangerous

# 创建缓存专用用户（只能操作 cache: 前缀的 key）
ACL SETUSER cache_user on >cache_pass ~cache:* +@all -keys -flushall -flushdb -config

# 创建管理员
ACL SETUSER admin on >admin_secret ~* &* +@all

# 禁用用户
ACL SETUSER readonly_user off

ACL CAT
# 1) "keyspace"
# 2) "read"
# 3) "write"
# 4) "set"
# 5) "sortedset"
# 6) "list"
# 7) "hash"
# 8) "string"
# 9) "bitmap"
# 10) "hyperloglog"
# 11) "geo"
# 12) "stream"
# 13) "pubsub"
# 14) "admin"
# 15) "fast"
# 16) "slow"
# 17) "blocking"
# 18) "dangerous"
# 19) "connection"
# 20) "transaction"
# 21) "scripting"

# 查看某分类下有哪些命令
ACL CAT dangerous
# 1) "flushdb"
# 2) "acl"
# 3) "slowlog"
# 4) "debug"
# 5) "role"
# 6) "keys"
# 7) "pfselftest"
# 8) "client"
# ...

# 删除用户
ACL DELUSER readonly_user

# 生成安全密码
ACL GENPASS
# "5d60k28ea..."

# 查看 ACL 拒绝日志
ACL LOG
# 1)  1) "count"
#     2) (integer) 5
#     3) "reason"
#     4) "command"
#     5) "context"
#     6) "toplevel"
#     7) "object"
#     8) "flushall"
#     9) "username"
#    10) "cache_user"
#    11) "age-seconds"
#    12) "120.5"
#    13) "client-info"
#    14) "id=55555 addr=127.0.0.1:42112 ..."

# 清空 ACL 日志（需要权限）
ACL LOG RESET

# 持久化 ACL 到文件 / 从文件加载
ACL SAVE
ACL LOAD

# redis.conf
aclfile /etc/redis/users.acl

# redis.conf
user default off
user admin on >admin_secret ~* &* +@all
aclfile /etc/redis/users.acl

# 禁用 default 用户（安全加固）
user default off

# 管理员：全部权限
user admin on >admin_secret ~* &* +@all

# 只读用户：只能读，不能写和执行危险命令
user app_readonly on >readonly_pass ~* +@read -@write -@dangerous

# 缓存服务：只能操作 cache:* 前缀
user app_cache on >cache_pass ~cache:* +@all -keys -flushall -flushdb -config

# 监控用户：无数据访问，仅监控
user monitor on >monitor_pass +ping +info +slowlog +client +cluster +latency

# 客户端开启追踪
CLIENT TRACKING ON

CLIENT TRACKING ON BCAST PREFIX user:

# 开启默认模式追踪
CLIENT TRACKING ON

# 开启广播模式追踪（仅接收 user: 前缀的失效通知）
CLIENT TRACKING ON BCAST PREFIX user:

# 开启追踪 + 指定重定向客户端 ID（失效消息发送到另一个连接）
CLIENT TRACKING ON REDIRECT 12345

# 关闭追踪
CLIENT TRACKING OFF

# 查看追踪状态
CLIENT TRACKINGINFO
# 1) "flags"
# 2) "on bcast"
# 3) "redirect"
# 4) (integer) 0
# 5) "prefixes"
# 6) 1) "user:"

# 使用 RESP3 连接
redis-cli -3

# 开启追踪
CLIENT TRACKING ON

# 开启追踪，不重定向（RESP3 自动区分）
CLIENT TRACKING ON OPTIN

// go-redis v9 支持 RESP3, 但客户端缓存需要自行实现
// 以下为概念性示例

rdb := redis.NewClient(&redis.Options{
    Addr:            "localhost:6379",
    Protocol:        3,        // 使用 RESP3
    DisableIdentity: true,     // 某些版本的兼容设置
})

// 开启追踪
rdb.Do(ctx, "CLIENT", "TRACKING", "ON")

// 本地缓存层（需要自行实现）
type LocalCache struct {
    mu   sync.RWMutex
    data map[string]string
}

func (c *LocalCache) Get(key string) (string, bool) {
    c.mu.RLock()
    defer c.mu.RUnlock()
    v, ok := c.data[key]
    return v, ok
}

func (c *LocalCache) Del(key string) {
    c.mu.Lock()
    defer c.mu.Unlock()
    delete(c.data, key)
}

// 读取时优先本地缓存，miss 时回源 Redis
func GetWithCache(ctx context.Context, cache *LocalCache, rdb *redis.Client, key string) (string, error) {
    // 1. 先查本地缓存
    if val, ok := cache.Get(key); ok {
        return val, nil
    }
    // 2. 回源 Redis
    val, err := rdb.Get(ctx, key).Result()
    if err != nil {
        return "", err
    }
    // 3. 写入本地缓存（带 TTL 的逻辑需自行实现）
    // cache.Set(key, val, ttl)
    return val, nil
}