Redis的Keyspace notifications功能

背景 项目需要引进一个缓存库, 对设备离线进行超时处理, 决定使用redis过期事件.

• 在Redis 2.8.0版本起，加入了“Keyspace notifications”（即“键空间通知”）的功能。

启用keyspace notifications功能

Keyspace notifications 功能默认是关闭的（默认地，Keyspace 时间通知功能是禁用的，因为它或多或少会使用一些CPU的资源），我们需要打开它。打开的方法也很简单，配置属性：notify-keyspace-events

$ vi /etc/redis/redis.conf

notify-keyspace-events “” 默认是空值, 将其设置为以下内容将打开这个功能.

#  K     Keyspace events, published with __keyspace@__ prefix.
#  E     Keyevent events, published with __keyevent@__ prefix.
#  g     Generic commands (non-type specific) like DEL, EXPIRE, RENAME, ...
#  $     String commands
#  l     List commands
#  s     Set commands
#  h     Hash commands
#  z     Sorted set commands
#  x     Expired events (events generated every time a key expires)
#  e     Evicted events (events generated when a key is evicted for maxmemory)
#  A     Alias for g$lshzxe, so that the "AKE" string means all the events.

将其设置为

notify-keyspace-events "KEx"

重启redis生效

测试Demo

redis-test.go

package main

import (
    "fmt"
    "time"
    "unsafe"

    log "github.com/astaxie/beego/logs"
    "github.com/gomodule/redigo/redis"
)

type PSubscribeCallback func(pattern, channel, message string)

type PSubscriber struct {
    client redis.PubSubConn
    cbMap  map[string]PSubscribeCallback
}

func (c *PSubscriber) PConnect(ip string, port uint16) {
    conn, err := redis.Dial("tcp", "127.0.0.1:6379")
    if err != nil {
        log.Critical("redis dial failed.")
    }

    c.client = redis.PubSubConn{conn}
    c.cbMap = make(map[string]PSubscribeCallback)

    go func() {
        for {
            log.Debug("wait...")
            switch res := c.client.Receive().(type) {
            case redis.Message:
                pattern := (*string)(unsafe.Pointer(&res.Pattern))
                channel := (*string)(unsafe.Pointer(&res.Channel))
                message := (*string)(unsafe.Pointer(&res.Data))
                c.cbMap[*channel](*pattern, *channel, *message)
            case redis.Subscription:
                fmt.Printf("%s: %s %dn", res.Channel, res.Kind, res.Count)
            case error:
                log.Error("error handle...")
                continue
            }
        }
    }()

}
func (c *PSubscriber) Psubscribe(channel interface{}, cb PSubscribeCallback) {
    err := c.client.PSubscribe(channel)
    if err != nil {
        log.Critical("redis Subscribe error.")
    }

    c.cbMap[channel.(string)] = cb
}

func timeoutCallback(patter, chann, device string) {
    log.Debug("timeoutCallback patter : "+patter+" channel : ", chann, " offline device : ", device)
}

func main() {
    var psub PSubscriber
    psub.PConnect("127.0.0.1", 6397)
    psub.Psubscribe("__keyevent@0__:expired", timeoutCallback)
    for {
        time.Sleep(1 * time.Second)
    }
}

1. 启动测试客户端

$ go run redis-test.go 

2022/05/08 16:03:54.563 [D]  wait...
__keyevent@0__:expired: psubscribe 1
2022/05/08 16:03:54.564 [D]  wait...
2022/05/08 16:04:22.966 [D]  timeoutCallback patter : __keyevent@0__:expired channel :  __keyevent@0__:expired  offline device :  ID19000101
2022/05/08 16:04:22.966 [D]  wait...

2. 在redis添加超时key/value数据

127.0.0.1:6379> set ID19000101 123 ex 5
OK

3. 结果

2022/05/08 16:03:54.563 [D]  wait...
__keyevent@0__:expired: psubscribe 1
2022/05/08 16:03:54.564 [D]  wait...
2022/05/08 16:04:22.966 [D]  timeoutCallback patter : __keyevent@0__:expired channel :  __keyevent@0__:expired  offline device :  ID19000101
2022/05/08 16:04:22.966 [D]  wait...