消息通知
Redis 中可以使用链表来实现消息通知功能。链表使用 LPUSH
和 RPOP
命令实现消息的入队和出队操作。但是有个问题,就是当链表中没有数据时调用 RPOP
命令会立刻失败返回,客户端无法阻塞等待。为了解决这个问题,可以使用 BRPOP
命令,当链表中没有数据时,客户端会阻塞等待。
BRPOP 命令使用示例
BRPOP key [key ...] timeout
key
:要阻塞等待的键。timeout
:超时时间,单位为秒。如果填写0
,则会一直阻塞。
示例
package mainimport ("context""fmt""time""github.com/go-redis/redis/v8"
)func main() {// 创建 Redis 客户端ctx := context.Background()rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379",Password: "", // no password setDB: 0, // use default DB})// 向队列中添加消息err := rdb.LPush(ctx, "myqueue", "message1").Err()if err != nil {panic(err)}// 使用 BRPOP 阻塞等待消息result, err := rdb.BRPop(ctx, 0, "myqueue").Result()if err != nil {panic(err)}// 输出结果fmt.Printf("Received message from %s: %s\n", result[0], result[1])
}
优先级队列
假设你的博客网站,每次有新文章时需要发送邮件通知,但是邮件发送需要一定的时间,如果采用 FIFO 的方式,可能会导致用户发送邮件的请求时间过长。同时每次有人进行注册都需要发送给注册用户一个确认邮件,如果用户注册时刚好也在发布文章,就会导致注册确认邮件等待很久才能发送,很显然这样的网站用户很难接受。所以,我们可以使用优先级队列,在队列中存储用户发送邮件的请求,并根据请求的优先级进行排序,优先级高的先发送。BRPOP
可以接受多个键,但是这些键之间并不是等价的,越靠近 BRPOP
的键优先级越高,借此特性可以实现区分优先级的任务队列。
优先级队列示例
package mainimport ("context""fmt""time""github.com/go-redis/redis/v8"
)func main() {// 创建 Redis 客户端ctx := context.Background()rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379",Password: "", // no password setDB: 0, // use default DB})// 向不同优先级的队列中添加消息err := rdb.LPush(ctx, "queue:1", "high-priority-message").Err()if err != nil {panic(err)}err = rdb.LPush(ctx, "queue:2", "medium-priority-message").Err()if err != nil {panic(err)}err = rdb.LPush(ctx, "queue:3", "low-priority-message").Err()if err != nil {panic(err)}// 使用 BRPOP 阻塞等待多个队列的消息result, err := rdb.BRPop(ctx, 0, "queue:1", "queue:2", "queue:3").Result()if err != nil {panic(err)}// 输出结果fmt.Printf("Received message from %s: %s\n", result[0], result[1])
}
发布/订阅 模式
除了实现任务队列外,Redis 还提供了一组命令可以让开发者实现“发布/订阅”(publish/subscribe)模式。
“发布/订阅”模式中包含两种角色,分别是发布者和订阅者。订阅者可以订阅一个或若干个频道(channel),而发布者可以向指定的频道发送消息,所有订阅此频道的订阅者都会收到此消息。
- 发出去的消息不会被持久化,也就是说当有客户端订阅
channel.1
后只能收到后续发布到该频道的消息,之前发送的就收不到了。 - 执行
SUBSCRIBE
命令后客户端会进入订阅状态,处于此状态下客户端不能使用除SUBSCRIBE
、UNSUBSCRIBE
、PSUBSCRIBE
和PUNSUBSCRIBE
这 4 个属于“发布/订阅”模式的命令之外的命令,否则会报错。 - 进入订阅状态后客户端可能收到 3 种类型的回复。每种类型的回复都包含 3 个值,第一个值是消息的类型,根据消息类型的不同,第二、三个值的含义也不同。消息类型可能的取值有以下 3 个:
subscribe
。表示订阅成功后的反馈信息。第二个值是订阅成功的频道名称,第三个值是当前客户端订阅的频道数量。message
。这个类型的回复是我们最关心的,它表示接收到的消息。第二个值表示产生消息的频道名称,第三个值是消息的内容。unsubscribe
。表示成功取消订阅某个频道。第二个值是对应的频道名称,第三个值是当前客户端订阅的频道数量,当此值为 0 时客户端会退出订阅状态,之后就可以执行其他非“发布/订阅”模式的命令了。
订阅与发布消息示例
package mainimport ("context""fmt""time""github.com/go-redis/redis/v8"
)func main() {// 创建 Redis 客户端ctx := context.Background()rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379",Password: "", // no password setDB: 0, // use default DB})// 订阅频道pubsub := rdb.Subscribe(ctx, "channel")defer pubsub.Close()// 启动一个 goroutine 来接收消息go func() {for {msg, err := pubsub.ReceiveMessage(ctx)if err != nil {panic(err)}fmt.Printf("Received message from %s: %s\n", msg.Channel, msg.Payload)}}()// 发布消息err := rdb.Publish(ctx, "channel", "Hello, Redis!").Err()if err != nil {panic(err)}// 等待一段时间以便接收消息time.Sleep(2 * time.Second)
}
按照规则订阅消息
除了可以使用 SUBSCRIBE
命令订阅指定名称的频道外,还可以使用 PSUBSCRIBE
命令订阅指定的规则。
按照规则订阅消息示例
package mainimport ("context""fmt""time""github.com/go-redis/redis/v8"
)func main() {// 创建 Redis 客户端ctx := context.Background()rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379",Password: "", // no password setDB: 0, // use default DB})// 订阅频道规则pubsub := rdb.PSubscribe(ctx, "channel.*")defer pubsub.Close()// 启动一个 goroutine 来接收消息go func() {for {msg, err := pubsub.ReceiveMessage(ctx)if err != nil {panic(err)}fmt.Printf("Received message from %s: %s\n", msg.Channel, msg.Payload)}}()// 发布消息err := rdb.Publish(ctx, "channel.1", "Hello, Redis!").Err()if err != nil {panic(err)}// 等待一段时间以便接收消息time.Sleep(2 * time.Second)
}
取消订阅
PUNSUBSCRIBE
命令用于取消订阅。如果没有参数则会退订所有规则。
取消订阅示例
package mainimport ("context""fmt""time""github.com/go-redis/redis/v8"
)func main() {// 创建 Redis 客户端ctx := context.Background()rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379",Password: "", // no password setDB: 0, // use default DB})// 订阅频道规则pubsub := rdb.PSubscribe(ctx, "channel.*")defer pubsub.Close()// 启动一个 goroutine 来接收消息go func() {for {msg, err := pubsub.ReceiveMessage(ctx)if err != nil {panic(err)}fmt.Printf("Received message from %s: %s\n", msg.Channel, msg.Payload)}}()// 发布消息err := rdb.Publish(ctx, "channel.1", "Hello, Redis!").Err()if err != nil {panic(err)}// 等待一段时间以便接收消息time.Sleep(2 * time.Second)// 取消订阅err = rdb.PUnsubscribe(ctx, "channel.*").Err()if err != nil {panic(err)}// 取消订阅后,客户端可以继续执行其他命令err = rdb.Set(ctx, "key", "value", 0).Err()if err != nil {panic(err)}fmt.Println("Set command executed successfully")
}