關于redigo中PubSub的一點小坑分析
更新時間：2019年01月03日 10:41:02 作者：Chen Jiehua
這篇文章主要給大家介紹了關于redigo中PubSub的一點小坑的相關資料，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧
前言
最近在用 golang 做一些 redis 相關的操作，選用了 redigo 這個第三方庫。然后在使用 Pub/Sub 的時候，卻發(fā)現(xiàn)了一個小坑……
Redis Client
首先，我們來初始化一個帶連接池的 Redis Client：
import (
"github.com/gomodule/redigo/redis"
)
type RedisClient struct {
pool *redis.Pool
}
func NewRedisClient(addr string, db int, passwd string) *RedisClient {
pool := &redis.Pool{
MaxIdle: 10,
IdleTimeout: 300 * time.Second,
Dial: func() (redis.Conn, error) {
c, err := redis.Dial("tcp", addr, redis.DialPassword(passwd), redis.DialDatabase(db))
if err != nil {
return nil, err
}
return c, nil
},
TestOnBorrow: func(c redis.Conn, t time.Time) error {
if time.Since(t) < time.Minute {
return nil
}
_, err := c.Do("PING")
return err
},
}
log.Printf("new redis pool at %s", addr)
client := &RedisClient{
pool: pool,
}
return client
}
Publish
然后我們可以簡單的實現(xiàn)一個 publish 方法：
func (r *RedisClient) Publish(channel, message string) (int, error) {
c := r.pool.Get()
defer c.Close()
n, err := redis.Int(c.Do("PUBLISH", channel, message))
if err != nil {
return 0, fmt.Errorf("redis publish %s %s, err: %v", channel, message, err)
}
return n, nil
}
Subscribe
接下來就是一個稍微復雜點的帶有心跳的 subscribe 方法：
func (r *RedisClient) Subscribe(ctx context.Context, consume ConsumeFunc, channel ...string) error {
psc := redis.PubSubConn{Conn: r.pool.Get()}
defer psc.Close()
log.Printf("redis pubsub subscribe channel: %v", channel)
if err := psc.Subscribe(redis.Args{}.AddFlat(channel)...); err != nil {
return err
}
done := make(chan error, 1)
// start a new goroutine to receive message
go func() {
for {
switch msg := psc.Receive().(type) {
case error:
done <- fmt.Errorf("redis pubsub receive err: %v", msg)
return
case redis.Message:
if err := consume(msg); err != nil {
done <- err
return
}
case redis.Subscription:
if msg.Count == 0 {
// all channels are unsubscribed
done <- nil
return
}
}
}
}()
// health check
tick := time.NewTicker(time.Minute)
defer tick.Stop()
for {
select {
case <-ctx.Done():
if err := psc.Unsubscribe(); err != nil {
return fmt.Errorf("redis pubsub unsubscribe err: %v", err)
}
return nil
case err := <-done:
return err
case <-tick.C:
if err := psc.Ping(""); err != nil {
return err
}
}
}
return nil
}
最后，我們寫一個簡單地 main 函數(shù)來調用 publish & subscribe：
func (r *RedisClient) Subscribe(ctx context.Context, consume ConsumeFunc, channel ...string) error {
psc := redis.PubSubConn{Conn: r.pool.Get()}
defer psc.Close()
log.Printf("redis pubsub subscribe channel: %v", channel)
if err := psc.Subscribe(redis.Args{}.AddFlat(channel)...); err != nil {
return err
}
done := make(chan error, 1)
// start a new goroutine to receive message
go func() {
for {
switch msg := psc.Receive().(type) {
case error:
done <- fmt.Errorf("redis pubsub receive err: %v", msg)
return
case redis.Message:
if err := consume(msg); err != nil {
done <- err
return
}
case redis.Subscription:
if msg.Count == 0 {
// all channels are unsubscribed
done <- nil
return
}
}
}
}()
// health check
tick := time.NewTicker(time.Minute)
defer tick.Stop()
for {
select {
case <-ctx.Done():
if err := psc.Unsubscribe(); err != nil {
return fmt.Errorf("redis pubsub unsubscribe err: %v", err)
}
return nil
case err := <-done:
return err
case <-tick.C:
if err := psc.Ping(""); err != nil {
return err
}
}
}
return nil
}
坑
咋一看之下，好像并沒有什么異常？然而，如果我們這時候去看 redis 的 tcp 連接，就可以發(fā)現(xiàn)一些貓膩：
$sudo netstat -antp | grep redis tcp 0 0 0.0.0.0:6379 0.0.0.0:* LISTEN 940/redis-server 0. tcp 0 0 172.16.8.128:6379 172.16.8.1:55010 ESTABLISHED 940/redis-server 0. tcp 0 0 172.16.8.128:6379 172.16.8.1:55015 ESTABLISHED 940/redis-server 0. tcp 0 0 172.16.8.128:6379 172.16.8.1:55009 ESTABLISHED 940/redis-server 0. tcp 0 0 172.16.8.128:6379 172.16.8.1:55005 ESTABLISHED 940/redis-server 0. tcp 0 0 172.16.8.128:6379 172.16.8.1:55012 ESTABLISHED 940/redis-server 0. tcp 0 0 172.16.8.128:6379 172.16.8.1:55011 ESTABLISHED 940/redis-server 0. tcp 0 0 172.16.8.128:6379 172.16.8.1:55013 ESTABLISHED 940/redis-server 0. tcp 0 0 172.16.8.128:6379 172.16.8.1:55007 ESTABLISHED 940/redis-server 0. tcp 0 0 172.16.8.128:6379 172.16.8.1:55006 ESTABLISHED 940/redis-server 0. tcp 0 0 172.16.8.128:6379 172.16.8.1:55014 ESTABLISHED 940/redis-server 0. tcp 0 0 172.16.8.128:6379 172.16.8.1:54972 ESTABLISHED 940/redis-server 0.
竟然是每一次 subscribe 就新建了一個連接，而 connection pool 似乎沒有什么作用。
更進一步地調試，我們發(fā)現(xiàn)在 defer psc.Close() 的時候就卡住了，也就是上面的 10 個 goroutine 其實并沒有正常退出。
Concurrent
排查許久之后，終于定位到了問題！引用 redigo 的說明：
Connections support one concurrent caller to the Receive method and one concurrent caller to the Send and Flush methods. No other concurrency is supported including concurrent calls to the Do method.
For full concurrent access to Redis, use the thread-safe Pool to get, use and release a connection from within a goroutine. Connections returned from a Pool have the concurrency restrictions described in the previous paragraph.
也就是說，雖然一個連接可以在不同的 goroutine 并發(fā)調用 Receive() 和 Subscribe()（subscribe調用了send和flush） ，但是卻不能再有其他并發(fā)操作（比如 Close()）。
其他相似的問題還可以參考 issue
Fix
知道了上面的原因之后，我們稍微修改一下 defer psc.Close() 的位置即可解決問題：
// start a new goroutine to receive message
go func() {
// IMPORTANT!
defer psc.Close()
for {
switch msg := psc.Receive().(type) {
case error:
總結
以上就是這篇文章的全部內容了，希望本文的內容對大家的學習或者工作具有一定的參考學習價值，如果有疑問大家可以留言交流，謝謝大家對腳本之家的支持。
相關文章
Redis是一種基于內存的數(shù)據(jù)存儲系統(tǒng),被廣泛用于解決高并發(fā)問題,下面這篇文章主要給大家介紹了關于Redis高并發(fā)超賣問題解決方案的相關資料,文中通過代碼介紹的非常詳細,需要的朋友可以參考下2024-02-02
這篇文章主要介紹了Redis開啟鍵空間通知實現(xiàn)超時通知的步驟,本文給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2020-06-06
本文介紹了Redis中常用的數(shù)據(jù)結構,包括字符串、列表、集合、哈希表、有序集合和Bitmap,并詳細說明了它們在各種場景下的使用,需要的朋友可以參考下2024-05-05
Redis Streams是Redis 5.0新增的數(shù)據(jù)類型,提供了一種日志結構化數(shù)據(jù)存儲方式,這種類型適合用于構建消息隊列、事件日志和處理時間序列數(shù)據(jù)的應用,本文介紹Redis 數(shù)據(jù)類型Streams相關知識,感興趣的朋友一起看看吧2024-10-10
這篇文章主要為大家介紹了Redis處理高并發(fā)之布隆過濾器詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪2022-12-12
本文主要介紹了Redis遠程連接Redis客戶端的實現(xiàn)步驟，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2022-06-06
這篇文章主要介紹了關于Redis中bitmap的原理和使用詳解,BitMap即位圖，使用每個位表示某種狀態(tài)，適合處理整型的海量數(shù)據(jù),本質上是哈希表的一種應用實現(xiàn),需要的朋友可以參考下2023-05-05
這篇文章主要介紹了淺談Redis哨兵模式的使用，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2020-12-12
為了保證我們線上服務的并發(fā)性和安全性，目前我們的服務一般拋棄了單體應用，采用的都是擴展性很強的分布式架構，這篇文章主要介紹了使用Redis實現(xiàn)分布式鎖的方法,需要的朋友可以參考下2022-06-06
這篇文章主要介紹了redis熱key問題怎樣解決,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2024-04-04