登录
当前位置： 首页 > 工具软件 > Go-redis-server > 使用案例 >

go使用go-redis操作redis(三)连接类型，pipline， 发布订阅

堵茂勋
2023-12-01

内容：

一 . 客户端Client（普通模式，主从模式，哨兵模式）
二. conn连接(连接， pipline， 发布订阅等)
三. 示例程序(连接， pipline， 发布订阅等)
客户端
  1. Client 普通模式的客户端

    go redis依据用途提供了多种客户端创建的函数， 如下：

    func NewClient(opt *Options) *Client
func NewFailoverClient(failoverOpt *FailoverOptions) *Client
func (c *Client) Context() context.Context
func (c *Client) Do(args ...interface{}) *Cmd
func (c *Client) DoContext(ctx context.Context, args ...interface{}) *Cmd
func (c *Client) Options() *Options
func (c *Client) PSubscribe(channels ...string) *PubSub
func (c *Client) Pipeline() Pipeliner
func (c *Client) Pipelined(fn func(Pipeliner) error) ([]Cmder, error)
func (c *Client) PoolStats() *PoolStats
func (c *Client) Process(cmd Cmder) error
func (c *Client) ProcessContext(ctx context.Context, cmd Cmder) error
func (c *Client) SetLimiter(l Limiter) *Client
func (c *Client) Subscribe(channels ...string) *PubSub
func (c *Client) TxPipeline() Pipeliner
func (c *Client) TxPipelined(fn func(Pipeliner) error) ([]Cmder, error)
func (c *Client) Watch(fn func(*Tx) error, keys ...string) error
func (c *Client) WithContext(ctx context.Context) *Client

    NewClient 创建一个普通连接

    NewFailoverClient 具有故障检测以及故障转移的client

    PSubscribe / Subscribe 发布订阅模式的client

    Pipeline 启用pipline管道模式的client

    PoolStats 连接池状态

    Close 关闭连接

  2. 集群模式的ClusterClient
    func NewClusterClient(opt *ClusterOptions) *ClusterClient
func (c *ClusterClient) Close() error
func (c *ClusterClient) Context() context.Context
func (c *ClusterClient) DBSize() *IntCmd
func (c *ClusterClient) Do(args ...interface{}) *Cmd
func (c *ClusterClient) DoContext(ctx context.Context, args ...interface{}) *Cmd
func (c *ClusterClient) ForEachMaster(fn func(client *Client) error) error
func (c *ClusterClient) ForEachNode(fn func(client *Client) error) error
func (c *ClusterClient) ForEachSlave(fn func(client *Client) error) error
func (c *ClusterClient) Options() *ClusterOptions
func (c *ClusterClient) PSubscribe(channels ...string) *PubSub
func (c *ClusterClient) Pipeline() Pipeliner
func (c *ClusterClient) Pipelined(fn func(Pipeliner) error) ([]Cmder, error)
func (c *ClusterClient) PoolStats() *PoolStats
func (c *ClusterClient) Process(cmd Cmder) error
func (c *ClusterClient) ProcessContext(ctx context.Context, cmd Cmder) error
func (c *ClusterClient) ReloadState() error
func (c *ClusterClient) Subscribe(channels ...string) *PubSub
func (c *ClusterClient) TxPipeline() Pipeliner
func (c *ClusterClient) TxPipelined(fn func(Pipeliner) error) ([]Cmder, error)
func (c *ClusterClient) Watch(fn func(*Tx) error, keys ...string) error
func (c *ClusterClient) WithContext(ctx context.Context) *ClusterClient

    方法与client大致相同

  3. 哨兵SentinelClient
    func NewSentinelClient(opt *Options) *SentinelClient
func (c *SentinelClient) CkQuorum(name string) *StringCmd
func (c SentinelClient) Close() error
func (c *SentinelClient) Context() context.Context
func (c *SentinelClient) Failover(name string) *StatusCmd
func (c *SentinelClient) FlushConfig() *StatusCmd
func (c *SentinelClient) GetMasterAddrByName(name string) *StringSliceCmd
func (c *SentinelClient) Master(name string) *StringStringMapCmd
func (c *SentinelClient) Masters() *SliceCmd
func (c *SentinelClient) Monitor(name, ip, port, quorum string) *StringCmd
func (c *SentinelClient) PSubscribe(channels ...string) *PubSub
func (c *SentinelClient) Ping() *StringCmd
func (c *SentinelClient) Process(cmd Cmder) error
func (c *SentinelClient) ProcessContext(ctx context.Context, cmd Cmder) error
func (c *SentinelClient) Remove(name string) *StringCmd
func (c *SentinelClient) Reset(pattern string) *IntCmd
func (c *SentinelClient) Sentinels(name string) *SliceCmd
func (c *SentinelClient) Set(name, option, value string) *StringCmd
func (c *SentinelClient) Slaves(name string) *SliceCmd
func (c SentinelClient) String() string
func (c *SentinelClient) Subscribe(channels ...string) *PubSub
func (c *SentinelClient) WithContext(ctx context.Context) *SentinelClient

连接：

创建量客户端之后，需要与redis建立连接才能够发送请求进行使用

在连接层， 提供了很多封装后的函数

  1. 普通连接Conn

    例如：

    涉及很多函数， 函数名称基本可以通过redis的命令进行类比找到，不一一列举

    func (c Conn) Append(key, value string) *IntCmd
func (c Conn) Auth(password string) *StatusCmd
func (c Conn) BLPop(timeout time.Duration, keys ...string) *StringSliceCmd
func (c Conn) BRPop(timeout time.Duration, keys ...string) *StringSliceCmd
func (c Conn) BRPopLPush(source, destination string, timeout time.Duration) *StringCmd
func (c Conn) BZPopMax(timeout time.Duration, keys ...string) *ZWithKeyCmd
func (c Conn) BZPopMin(timeout time.Duration, keys ...string) *ZWithKeyCmd
func (c Conn) BgRewriteAOF() *StatusCmd
func (c Conn) BgSave() *StatusCmd
func (c Conn) BitCount(key string, bitCount *BitCount) *IntCmd
func (c Conn) BitField(key string, args ...interface{}) *IntSliceCmd
func (c Conn) BitOpAnd(destKey string, keys ...string) *IntCmd

....

func (c Conn) HKeys(key string) *StringSliceCmd
func (c Conn) HLen(key string) *IntCmd
func (c Conn) HMGet(key string, fields ...string) *SliceCmd
func (c Conn) HMSet(key string, fields map[string]interface{}) *StatusCmd
func (c Conn) HScan(key string, cursor uint64, match string, count int64) *ScanCmd
func (c Conn) HSet(key, field string, value interface{}) *BoolCmd
func (c Conn) HSetNX(key, field string, value interface{}) *BoolCmd
func (c Conn) HVals(key string) *StringSliceCmd
  2. Pipeline

    pipline 管道批量执行命令，可以节约带宽

    提供的方法与conn基本一致

    func (c Pipeline) Append(key, value string) *IntCmd
func (c Pipeline) Auth(password string) *StatusCmd
func (c Pipeline) BLPop(timeout time.Duration, keys ...string) *StringSliceCmd
func (c Pipeline) BRPop(timeout time.Duration, keys ...string) *StringSliceCmd
func (c Pipeline) BRPopLPush(source, destination string, timeout time.Duration) *StringCmd
func (c Pipeline) BZPopMax(timeout time.Duration, keys ...string) *ZWithKeyCmd
func (c Pipeline) BZPopMin(timeout time.Duration, keys ...string) *ZWithKeyCmd
func (c Pipeline) BgRewriteAOF() *StatusCmd

.....
  3. PubSub

    发布订阅模式

    func (c *PubSub) Channel() <-chan *Message
func (c *PubSub) ChannelSize(size int) <-chan *Message
func (c *PubSub) ChannelWithSubscriptions(size int) <-chan interface{}
func (c *PubSub) Close() error
func (c *PubSub) PSubscribe(patterns ...string) error
func (c *PubSub) PUnsubscribe(patterns ...string) error
func (c *PubSub) Ping(payload ...string) error
func (c *PubSub) Receive() (interface{}, error)
func (c *PubSub) ReceiveMessage() (*Message, error)
func (c *PubSub) ReceiveTimeout(timeout time.Duration) (interface{}, error)
func (c *PubSub) String() string
func (c *PubSub) Subscribe(channels ...string) error
func (c *PubSub) Unsubscribe(channels ...string) error

    可以建立channel Channel()

    设置超时时间ReceiveTimeout

    Subscribe订阅

    Unsubscribe取消订阅

    PSubscribe 发布消息等

  4. 排他 TX
    func (c Tx) Get(key string) *StringCmd
func (c Tx) GetBit(key string, offset int64) *IntCmd
func (c Tx) GetRange(key string, start, end int64) *StringCmd
func (c Tx) GetSet(key string, value interface{}) *StringCmd
func (c Tx) HDel(key string, fields ...string) *IntCmd
func (c Tx) HExists(key, field string) *BoolCmd
func (c Tx) HGet(key, field string) *StringCmd
func (c Tx) HGetAll(key string) *StringStringMapCmd
func (c Tx) HIncrBy(key, field string, incr int64) *IntCmd
func (c Tx) HIncrByFloat(key, field string, incr float64) *FloatCmd
func (c Tx) HKeys(key string) *StringSliceCmd

    在conn的基础上加入排他性功能

    提供的函数及方法与conn基本相同， 可以参考conn进行使用

示例程序

示例：

func ExampleClient() {
	redisdb := redis.NewClient(&redis.Options{
		Addr:     "192.168.137.18:6379",
		Password: "", // no password set
		DB:       0,  // use default DB
	})

	err := redisdb.Set("key", "value", 0).Err()
	if err != nil {
		panic(err)
	}

	val, err := redisdb.Get("key").Result()
	if err != nil {
		panic(err)
	}
	fmt.Println("key", val)

	val2, err := redisdb.Get("missing_key").Result()
	if err == redis.Nil {
		fmt.Println("missing_key does not exist")
	} else if err != nil {
		panic(err)
	} else {
		fmt.Println("missing_key", val2)
	}

}

func main() {
	ExampleClient()
}

输出：

key value
missing_key does not exist
  1. NewClient创建连接：
redisdb := redis.NewClient(&redis.Options{
    Addr:     "localhost:6379", // use default Addr
    Password: "",               // no password set
    DB:       0,                // use default DB
})

pong, err := redisdb.Ping().Result()
fmt.Println(pong, err)

如果使用tls/ssl 则在Options参数 TLSConfig *tls.Config 中指定

  1. 主从故障转移NewFailoverClient
    redisdb := redis.NewFailoverClient(&redis.FailoverOptions{
    MasterName:    "master",
    SentinelAddrs: []string{":26379"},
})
redisdb.Ping()
  2. 集群NewClusterClient
    redisdb := redis.NewClusterClient(&redis.ClusterOptions{
    Addrs: []string{":7000", ":7001", ":7002", ":7003", ":7004", ":7005"},
})
redisdb.Ping()
  3. url样式建立连接 ParseURL
    opt, err := redis.ParseURL("redis://:qwerty@localhost:6379/1")
if err != nil {
    panic(err)
}
fmt.Println("addr is", opt.Addr)
fmt.Println("db is", opt.DB)
fmt.Println("password is", opt.Password)

// Create client as usually.
_ = redis.NewClient(opt)
  4. pipline
    pipe := redisdb.Pipeline()

incr := pipe.Incr("pipeline_counter")
pipe.Expire("pipeline_counter", time.Hour)

// Execute
//
//     INCR pipeline_counter
//     EXPIRE pipeline_counts 3600
//
// using one redisdb-server roundtrip.
_, err := pipe.Exec()
fmt.Println(incr.Val(), err)

    Pipelined

    var incr *redis.IntCmd
_, err := redisdb.Pipelined(func(pipe redis.Pipeliner) error {
    incr = pipe.Incr("pipelined_counter")
    pipe.Expire("pipelined_counter", time.Hour)
    return nil
})
fmt.Println(incr.Val(), err)

    排他TxPipeline

    pipe := redisdb.TxPipeline()

incr := pipe.Incr("tx_pipeline_counter")
pipe.Expire("tx_pipeline_counter", time.Hour)

// Execute
//
//     MULTI
//     INCR pipeline_counter
//     EXPIRE pipeline_counts 3600
//     EXEC
//
// using one redisdb-server roundtrip.
_, err := pipe.Exec()
fmt.Println(incr.Val(), err)

    rdb := redis.NewClient(&redis.Options{
    Addr: ":6379",
})
rdb.AddHook(redisHook{})

rdb.Pipelined(func(pipe redis.Pipeliner) error {
    pipe.Ping()
    pipe.Ping()
    return nil
})
  5. 发布订阅
    sub := client.Subscribe(queryResp)
iface, err := sub.Receive()
if err != nil {
    // handle error
}

// Should be *Subscription, but others are possible if other actions have been
// taken on sub since it was created.
switch iface.(type) {
case *Subscription:
    // subscribe succeeded
case *Message:
    // received first message
case *Pong:
    // pong received
default:
    // handle error
}

ch := sub.Channel()

    示例程序

    package main

import (
	"github.com/go-redis/redis"
	"time"
	"fmt"
)


func ExampleClient() {
	redisdb := redis.NewClient(&redis.Options{
		Addr:     "192.168.137.18:6379",
		Password: "", // no password set
		DB:       0,  // use default DB
	})
	//rdb.AddHook()

	pubsub := redisdb.Subscribe("mychannel1")

	// Wait for confirmation that subscription is created before publishing anything.
	_, err := pubsub.Receive()
	if err != nil {
		panic(err)
	}

	// Go channel which receives messages.
	ch := pubsub.Channel()

	// Publish a message.
	err = redisdb.Publish("mychannel1", "hello").Err()
	if err != nil {
		panic(err)
	}

	time.AfterFunc(time.Second, func() {
		// When pubsub is closed channel is closed too.
		_ = pubsub.Close()
	})

	// Consume messages.
	for msg := range ch {
		fmt.Println(msg.Channel, msg.Payload)
	}
}

func main() {
	ExampleClient()
}

    输出：

    mychannel1 hello

    示例2

    func ExampleClient2() {
	redisdb := redis.NewClient(&redis.Options{
		Addr:     "192.168.137.18:6379",
		Password: "", // no password set
		DB:       0,  // use default DB
	})
	//rdb.AddHook()
	pubsub := redisdb.Subscribe("mychannel2")
	defer pubsub.Close()

	for i := 0; i < 2; i++ {
		// ReceiveTimeout is a low level API. Use ReceiveMessage instead.
		msgi, err := pubsub.ReceiveTimeout(time.Second)
		if err != nil {
			break
		}

		switch msg := msgi.(type) {
		case *redis.Subscription:
			fmt.Println("subscribed to", msg.Channel)

			_, err := redisdb.Publish("mychannel2", "hello").Result()
			if err != nil {
				panic(err)
			}
		case *redis.Message:
			fmt.Println("received", msg.Payload, "from", msg.Channel)
		default:
			panic("unreached")
		}
	}

}

    输出：

    subscribed to mychannel2
received hello from mychannel2
 类似资料：

相关阅读

Redis 订阅发布_Jedis实现方法使用PHP连接到RedisRedis中的高效索引类型操作为什么Redis具有发布/订阅功能？Redis中五种数据类型简单操作

相关文章

Redis数据类型Redis数值操作命令Redis的数据类型详解和使用：List类型Go语言链表操作Redis的数据类型详解和使用：key、String类型

相关问答

无法使用Redis cli连接到AWS Redis终结点。连接超时Redis莴苣发布连接到Redis Sentinel集群时使用redis-py的MasterNotFoundErrorgo - Go连接Kafka的问题？无法使用jedis连接redis sentinel

相关文档

Go 实战开发Go RPC 开发指南用 Go 来做以太坊开发Redis 常用命令Redis 开发运维实践指南
快捷导航： 新手教程 算法原理 架构设计 Java进阶 数据库进阶 大厂专栏 面试经验 编程笔记 编程问答 所有专题 文档资料 工具软件 电子书籍 小牛导航
在线工具： 房贷计算器 个税计算器 Linux命令查询 Json格式化 正则表达式 颜色转换 AES加解密 SHA1加密 MD5加密 毒鸡汤 字数统计 随机密码生成 进制转换 Base64编解码 励志句子
Copyright © 2019-2024 小牛知识库@xnip.cn. All Rights Reserved.