redigo模块的详细使用

连接

Conn接口是redis操作过程中比较重要的接口。应用一般通过调用Dial、DialWithTimeout、NewConn函数来创建一个链接。
注意：应用必须在操作redis结束之后去调用该连接的Close方法来关闭连接，以防止资源消耗以及其他问题。

Conn接口下的相关方法

type Conn interface {    // 关闭链接    Close() error
    // 当链接不可用时返回非空值    Err() error
    //Do方法向redis服务端发送命令并返回接收到响应    Do(commandName string, args ...interface{}) (reply interface{}, err error)
    //Send 将相关的命令写入客户端的buffer中    Send(commandName string, args ...interface{}) error
    // Flush 将客户端的输出缓冲内容刷新到redis服务器端    Flush() error
    // Receive 从redis服务端接收单个回复    Receive() (reply interface{}, err error)}

常用的创建Conn的方式

# 使用Dial函数# func Dial(network, address string, options ...DialOption) (Conn, error)/*可选的参数func DialConnectTimeout(d time.Duration) DialOptionfunc DialDatabase(db int) DialOptionfunc DialKeepAlive(d time.Duration) DialOptionfunc DialNetDial(dial func(network, addr string) (net.Conn, error)) DialOptionfunc DialPassword(password string) DialOptionfunc DialReadTimeout(d time.Duration) DialOptionfunc DialWriteTimeout(d time.Duration) DialOption*/c,err := redis.Dial("tcp","localhost:32771")defer c.Close()


# 使用DialTimeout函数(默认增加了连接、读写超时时间)// func DialTimeout(network, address string, connectTimeout, readTimeout, writeTimeout time.Duration) (Conn, error)

# 使用NewConn函数(将一个网络连接转换成redis连接)// func NewConn(netConn net.Conn, readTimeout, writeTimeout time.Duration) Conn

redis命令的执行

在上面Conn的接口中可以看到有一个Do方法，可以用来执行redis命令操作。

Do方法定义

//使用Conn的Do方法来操作redis命令Do(commandName string, args ...interface{}) (reply interface{}, err error)
//示例n, err := conn.Do("APPEND", "key", "value")

redis连接和命令执行的简单示例

简单示例：

$ cat redis-conn.gopackage mainimport (    "fmt"    "os"    "time"    redis "github.com/gomodule/redigo/redis")func main() {    //使用redis封装的Dial进行tcp连接.设置长连接时间为1s,连接超时时间为5s,读写超时均为1s，并设置密码进行连接    c,err := redis.Dial("tcp","localhost:32771",redis.DialKeepAlive(1*time.Second),redis.DialPassword("123qweasd"),redis.DialConnectTimeout(5*time.Second),redis.DialReadTimeout(1*time.Second),redis.DialWriteTimeout(1*time.Second))    errCheck(err)
    defer c.Close()
    //使用redis的string类型获取set的k/v信息    r,getErr := redis.String(c.Do("get","url"))    errCheck(getErr)    fmt.Println(r)
}
func errCheck(err error) {    if err != nil {        fmt.Println("sorry,has some error:",err)        os.Exit(-1)        }    }
$ go run redis-conn.goxxbandy.github.io
# 由于redis服务器设置了密码，如果密码错误会报如下异常$ go run redis-conn.gosorry,has some error: ERR invalid passwordexit status 255

使用Pipelining操作redis

注意:建议使用redis自身包含的命令进行批量操作而不是使用pipelining，比如mset、mget、hmset、hmget等等。原子性可能会更好一些

Conn会使用Send、Flush、Receive来支持pipeline操作。

Send(commandName string, args ...interface{}) errorFlush() errorReceive() (reply interface{}, err error)

Send会写入命令到连接的输出缓冲区里。Flush会将输出缓冲区中的数据刷新到服务端。Receive回去荣服务端读取单个响应。

c.Send("SET", "foo", "bar")c.Send("GET", "foo")c.Flush()c.Receive()v, err = c.Receive()

其实，在Do方法中会包含Send/Flush/Receive等方法。

使用Send和Do方法来实现pipeline。

c.Send("MULTI")c.Send("INCR", "foo")c.Send("INCR", "bar")r, err := c.Do("EXEC")fmt.Println(r) //[1,1]

并发

一般并发访问redis，建议使用安全线程池去获取连接，从一个goroutine中使用和释放一个链接。正如前面提到的，从线程池中获取的连接会有并发限制。

发布/订阅模式(Publish and Subscribe)

使用Send/Flush/Receive方法去实现Pub/Sub订阅。

c.Send("SUBSCRIBE","ANSIBLE-KEY")c.Flush()for {    reply,err := c.Receive()    if err != nil {        return err    }    //process pushed message}

PubSubConn类型使用convenience方法包装了一个Conn来实现了订阅。Subscribe/PSubscribe/Unsubscribe/PUnsubscribe方法会send并且flush一个订阅管理命令。接收方法会将push的消息转换成一个可以进行switch的合适类型。

psc := redis.PubSubConn{Conn: c}
psc.Subscribe("example")for {    switch v := psc.Receive().(type) {    case redis.Message:        fmt.Printf("%s: message: %s\n", v.Channel, v.Data)    case redis.Subscription:        fmt.Printf("%s: %s %d\n", v.Channel, v.Kind, v.Count)    case error:        return v    }}

Redigo操作redis的几种场景示例

使用`Args`接口数组操作处理批量操作

使用`String和Int`获取字符串类型的输出

也就是set、get、mset、mget之类的命令。
string相关命令^[4]

注意:当获取数字类型的key时需要使用redis.Int()获取数字类型的输出结果

$ cat redis-string.go....func main() {    c,connErr := redisConn("localhost","32771","123qweasd")    errCheck("connErr",connErr)
    getV,_ := redis.String(c.Do("get","name"))    getV2,_ := redis.Int(c.Do("get","id"))    fmt.Println(getV,getV2)}$ go run redis-string.goxxb 1214
127.0.0.1:6379> mget name id1) "xxb"2) "1214"

mset、mget、expire之类的使用：

// core functions_,err = c.Do("mset","id",100,"fn",200)
_,expireErr = c.Do("expire","id",10)
if r,err := redis.Ints(c.Do("mget","id","fn")); err == nil {    for _,v := range {        fmt.Println("value:",v)    }}

lpush、lpop、rpush、rpop等相关命令操作:

// core functions_,err = c.Do("lpush","updateid","0407","0408","0409")
r,err ：= redis.String(c.Do("lpop","updateid"))

使用`redis`的安全链连接池

使用连接池可以高效的管理redis的连接，并可以方便地控制redis的并发性能。

使用`redis`的发布订阅模式

相关的结构体以及方法：

//PubSubConn结构体type PubSubConn struct {    Conn    Conn}
//使用redis链接初始化一个pubsub链接psc := redis.PubSubConn{Conn:redis.conn}
//pubsub链接的相关方法//关闭pubsub链接func (c PubSubConn) Close() error//批量订阅(按模式订阅)func (c PubSubConn) PSubscribe(channel ...interface{}) error
//取消订阅func (c PubSubConn) PUnsubscribe(channel ...interface{}) error
// Receive返回消费的信息作为Subscription, Message, Pong or error.返回值建议直接在switch中使用func (c PubSubConn) Receive() interface{}type Subscription struct {    Kind    string  //订阅,取消订阅    Channel string  //被改变的channel    //订阅的链接数量    Count   int}type Message struct {    // 初始的channel    Channel string    // 匹配的模式    Pattern string    // channel的消息体    Data []byte}type


//订阅和取消订阅func (c PubSubConn) Subscribe(channel ...interface{}) errorfunc (c PubSubConn) Unsubscribe(channel ...interface{}) error

测试示例：


package main
import (    "context"    "fmt"    "time"
    "github.com/gomodule/redigo/redis")
//func listenPubSubChannels(ctx context.Context, redisServerAddr string,    onStart func() error,    onMessage func(channel string, data []byte) error,    channels ...string) error {
    //一分钟的心跳检测    const healthCheckPeriod = time.Minute
    //构建一个redis链接    c, err := redis.Dial("tcp", redisServerAddr,        redis.DialReadTimeout(healthCheckPeriod+10*time.Second),        redis.DialWriteTimeout(10*time.Second))    if err != nil {        return err    }    defer c.Close()
    //构建一个pubsub链接    psc := redis.PubSubConn{Conn: c}
    //订阅channels    if err := psc.Subscribe(redis.Args{}.AddFlat(channels)...); err != nil {        return err    }    //构造chan来检测通知状态    done := make(chan error, 1)
    //启动一个goroutine来接受来自server端的通知    go func() {        for {            //使用interface{}.(type) 来获取对应的类型，并借助switch和case进行interface{}的类型判断            switch n := psc.Receive().(type) {            case error:                done <- n                return            case redis.Message:                if err := onMessage(n.Channel, n.Data); err != nil {                    done <- err                    return                }            case redis.Subscription:                switch n.Count {                case len(channels):                    // Notify application when all channels are subscribed.                    if err := onStart(); err != nil {                        done <- err                        return                    }                case 0:                    // Return from the goroutine when all channels are unsubscribed.                    done <- nil                    return                }            }        }    }()
    ticker := time.NewTicker(healthCheckPeriod)    defer ticker.Stop()loop:    for err == nil {        select {        case <-ticker.C:            // Send ping to test health of connection and server. If            // corresponding pong is not received, then receive on the            // connection will timeout and the receive goroutine will exit.            if err = psc.Ping(""); err != nil {                break loop            }        case <-ctx.Done():            break loop        case err := <-done:            // Return error from the receive goroutine.            return err        }    }
    // Signal the receiving goroutine to exit by unsubscribing from all channels.    psc.Unsubscribe()
    // Wait for goroutine to complete.    return <-done}
func publish(redisServerAddr string) {    c, err := redis.Dial("tcp",redisServerAddr)    if err != nil {        fmt.Println(err)        return    }    defer c.Close()
    c.Do("PUBLISH", "c1", "hello")}
// This example shows how receive pubsub notifications with cancelation and// health checks.func main() {    redisServerAddr := "localhost:32771"
    ctx, cancel := context.WithCancel(context.Background())
    //ctx和start callback很好的解决了丢失的消息的填充(使用goroutine来占住message类型的消息)    listenErr := listenPubSubChannels(ctx,        redisServerAddr,        func() error {            // The start callback is a good place to backfill missed            // notifications. For the purpose of this example, a goroutine is            // started to send notifications.            go publish(redisServerAddr)            return nil        },        func(channel string, message []byte) error {            fmt.Printf("channel: %s, message: %s\n", channel, message)
            // For the purpose of this example, cancel the listener's context            // after receiving last message sent by publish().            if string(message) == "goodbye" {                cancel()            }            return nil        },        "ansible-key")
    if listenErr != nil {        fmt.Println(listenErr)        return    }
}

欢迎关注我的公众号 null

References

[1] redigo模块: https://github.com/gomodule/redigo
[2] redis-cluster客户端实现: https://github.com/mna/redisc/
[3] go-redis模块: https://github.com/go-redis/redis/
[4] string相关命令: http://redisdoc.com/string/index.html

ITPub博客

Golang操作Redis指南

相关模块以及安装方式

redigo模块的详细使用

连接

redis命令的执行

redis连接和命令执行的简单示例

使用Pipelining操作redis

并发

发布/订阅模式(Publish and Subscribe)

Redigo操作redis的几种场景示例

使用`Args`接口数组操作处理批量操作

使用`String和Int`获取字符串类型的输出

使用`redis`的安全链连接池

使用`redis`的发布订阅模式

References

最新文章

ITPub博客

Golang操作Redis指南

相关模块以及安装方式

redigo模块的详细使用

连接

redis命令的执行

redis连接和命令执行的简单示例

使用Pipelining操作redis

并发

发布/订阅模式(Publish and Subscribe)

Redigo操作redis的几种场景示例

使用Args接口数组操作处理批量操作

使用String和Int获取字符串类型的输出

使用redis的安全链连接池

使用redis的发布订阅模式

References

最新文章

使用`Args`接口数组操作处理批量操作

使用`String和Int`获取字符串类型的输出

使用`redis`的安全链连接池

使用`redis`的发布订阅模式