我正在使用 redigo 制作 PubSub，并且连接是由 redis 池创建的。
这是 Redis 池代码：
package main
import (
"os"
"os/signal"
"syscall"
"time"
"github.com/gomodule/redigo/redis"
)
type IRedis interface {
Addr() string
Conn() redis.Conn
Set(key string, body string) error
Close()
}
type Redis struct {
addr string
pool *redis.Pool
}
func NewRedis(addr string) *Redis {
r := &Redis{
addr,
&redis.Pool{
MaxIdle: 50000,
IdleTimeout: 240 * time.Second,
Dial: func() (redis.Conn, error) {
c, err := redis.Dial("tcp", addr)
if err != nil {
return nil, err
}
return c, err
},
TestOnBorrow: func(c redis.Conn, t time.Time) error {
_, err := c.Do("PING")
return err
},
},
}
r.cleanupHook()
return r
}
func (r *Redis) cleanupHook() {
c := make(chan os.Signal, 1)
signal.Notify(c, os.Interrupt)
signal.Notify(c, syscall.SIGTERM)
signal.Notify(c, syscall.SIGKILL)
go func() {
<-c
r.pool.Close()
os.Exit(0)
}()
}
func (r *Redis) Addr() string {
return r.addr
}
func (r *Redis) Conn() redis.Conn {
return r.pool.Get()
}
func (r *Redis) Set(key string, body string) error {
p := r.pool.Get()
defer p.Close()
_, err := p.Do("SET", key, body)
return err
}
func (r *Redis) Close() {
r.pool.Close()
}
这是我的 PubSub 代码：
package main
import (
"log"
"github.com/gomodule/redigo/redis"
)
type PubSub struct {
send chan string
conn *redis.PubSubConn
}
func NewPubSub(s chan string, c *redis.PubSubConn) (*PubSub, error) {
err := c.Subscribe("urls")
if err != nil {
return nil, err
}
return &PubSub{s, c}, nil
}
func (ps *PubSub) Start() {
for {
switch v := ps.conn.Receive().(type) {
case redis.Message:
data := string(v.Data)
ps.send <- data
case redis.Subscription:
log.Printf("subscription message: %s: %s %d\n", v.Channel, v.Kind, v.Count)
case error:
log.Println("error:", v)
return
}
}
}
func (ps *PubSub) Close() {
ps.conn.Close()
}
在我的 PubSub 频道中收到 10k 条消息后，连接丢失，我收到一个 redis.Error 消息 EOF
任何想法为什么会发生这种情况？即使我在本地运行，问题也会发生