^-^
【go分布式中间件】Redis
最近更新：2024-12-13   |   字数总计：1.7k   |   阅读估时：8分钟   |   阅读量：次
  1. Redis
    1. redigo基本命令操作
    2. redigo管道以及事务
      1. 管道
      2. 命令事务
      3. 脚本事务
    3. redigo订阅发布与stream

Redis

  • Redis通信模式：请求回应模式，client发送request，redis回复一个response
  • Redis通信协议：采用特殊字符\r\n来分割有意义的数据，也就是二进制安全字符串（用长度来标识字符串的长度），所以redis可以用来存储二进制数据，例如：
    1
    2
    3
    set key val
    # "*3\r\n$3\r\nSET\r\n$3\r\nkey\r\n$3\r\nval\r\n"
    # 去掉 "*3 $3 SET $3 key $3 val" *3标识后面有意义的段为3个，$3标识后面有意义的字符为3个
  • 协议传输过程中，是字节流，即[]byte；redis协议需要将整数、浮点数转换为字符串；

redigo基本命令操作

  • 启动redis
    1
    2
    3
    4
    5
    redis-server /home/redis_data/redis.conf
    # 关闭
    redis-cli -h 127.0.0.1 -a xxxxx shutdown
    # 查看redis情况
    ps aux | grep redis-server
  • basic.go
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89
    90
    91
    92
    93
    94
    95
    96
    97
    98
    99
    100
    101
    102
    103
    104
    105
    106
    107
    108
    109
    110
    111
    112
    113
    114
    115
    116
    117
    package main

    import (
    	"fmt"

    	"github.com/garyburd/redigo/redis"
    )

    func main() {
    	conn, err := redis.Dial("tcp", fmt.Sprintf("%s:%d", "127.0.0.1", 6379), redis.DialPassword("123456"))
    	if err != nil {
    		panic(err)
    	}
    	defer (func() {
    		fmt.Println("connection close")
    		conn.Close()
    	})()
    	if false {
    		conn.Do("set", "hello", "world")
    		// redis.String当我们明确知道返回的[]byte是string类型时，可以做转换
    		rpy, err := redis.String(conn.Do("get", "hello"))
    		if err != nil {
    			panic(err)
    		}
    		fmt.Println(rpy)
    	}
    	if false {
    		conn.Do("set", "hello2", 1)
    		// redis.Int当我们明确知道返回的[]byte是int类型时，可以做转换
    		rpy, err := redis.Int(conn.Do("get", "hello2"))
    		if err != nil {
    			panic(err)
    		}
    		fmt.Println(rpy)
    	}
    	if false {
    		// 通过封装好的args传参
    		args := redis.Args{}.Add("list").Add("kk").Add("ll").Add("ss")
    		// 直接传参
    		// conn.Do("lpush", "list", "mm", "dd", "kk")
    		conn.Do("lpush", args...)
    		rpy, err := redis.Strings(conn.Do("lrange", "list", 0, -1))
    		if err != nil {
    			panic(err)
    		}
    		fmt.Println(rpy)
    	}
    	if false {
    		conn.Do("del", "list")
    		// brpop是为空时会阻塞的命令, brpop list(key) 20(超时时间)
    		val, err := redis.Strings(conn.Do("brpop", redis.Args{}.Add("list").Add(20)...))
    		// 阻塞之后 redis-cli -a password
    		// lpush list ll 从redis客户端push一个数进去，就会退出阻塞
    		if err != redis.ErrNil {
    			fmt.Println(val, err)
    		}
    	}
    	if false {
    		conn.Do("del", "list")
    		// 将结果转换成一个map[string]string 类型
    		vals, err := redis.StringMap(conn.Do("brpop", redis.Args{}.Add("list").Add(20)...))
    		if err != redis.ErrNil {
    			fmt.Println(vals, err)
    		}
    	}
    	if false {
    		conn.Do("del", "list")
    		conn.Do("lpush", "list", "mm", 100)
    		// 当我们知道我们的值是不同的类型，就可以使用Values来接收，同时使用对应的值类型进行Scan
    		vals, _ := redis.Values(conn.Do("lrange", "list", 0, -1))
    		var name string
    		var score int
    		redis.Scan(vals, &score, &name)
    		fmt.Println(name, score)
    	}
    	if true {
    		var p1, p2 struct {
    			Name string `redis:"name"` // 使用tag规定写入redis的key是小写的
    			Age  string `redis:"age"`
    			Sex  string `redis:"sex"`
    		}
    		p1.Age = "18"
    		p1.Name = "mm"
    		p1.Sex = "male"
    		// addflat对结构体进行展开
    		// 展开后类似 HMSET role:1001 name "mm" age "18" sex "male"
    		args1 := redis.Args{}.Add("role:1001").AddFlat(&p1)
    		if _, err := conn.Do("hmset", args1...); err != nil {
    			fmt.Println(err)
    			return
    		}

    		m := map[string]string{
    			"name": "dd",
    			"age":  "20",
    			"sex":  "female",
    		}
    		args2 := redis.Args{}.Add("role:1002").AddFlat(m)
    		if _, err := conn.Do("hmset", args2...); err != nil {
    			fmt.Println(err)
    			return
    		}

    		for _, id := range []string{"role:1001", "role:1002"} {
    			v, err := redis.Values(conn.Do("HGETALL", id))
    			if err != nil {
    				fmt.Println(err)
    				return
    			}
    			if err := redis.ScanStruct(v, &p2); err != nil {
    				fmt.Println(err)
    				return
    			}
    			fmt.Printf("%+v\n", p2)
    		}
    	}
    }

redigo管道以及事务

管道

  • 管道：client同时按次序发送多个request，redis按照顺序执行后按顺序返回response
  • redigo的函数：
    1. Do == Send + Flush + Receive
    2. Send : 填充到客户端的output buffer
    3. Flush ： 刷一下output buffer内容到redis
    4. Receive ： 接收一个redis的返回
  • pipeline.go
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    package main

    import (
    	"fmt"
    	"math/rand"
    	"time"

    	"github.com/garyburd/redigo/redis"
    )

    func main() {
    	c, err := redis.Dial("tcp", fmt.Sprintf("%s:%d", "127.0.0.1", 6379), redis.DialPassword("123456"))
    	if err != nil {
    		panic(err)
    	}
    	defer (func() {
    		fmt.Println("connection close")
    		c.Close()
    	})()
    	if false {
    		c.Send("del", "set", "list", "zset")
    		c.Send("sadd", "set", "mm", "dd", "kk")
    		c.Send("lpush", "list", 10001, 10002, 10003)
    		c.Send("smembers", "set")
    		c.Send("lrange", "list", 0, -1)
    		c.Flush()
    		c.Receive() //del
    		c.Receive() //sadd
    		c.Receive() //1push

    		mbrs, err := redis.Strings(c.Receive())
    		if err != redis.ErrNil {
    			fmt.Println(mbrs)
    		}

    		lsts, err := redis.Strings(c.Receive())
    		if err != redis.ErrNil {

    			fmt.Println(lsts)
    		}
    	}
    	if false {
    		c.Send("del", "set", "list", "zset")
    		c.Send("sadd", "set", "mm", "dd", "kk")
    		c.Send("lpush", "list", 10001, 10002, 10003)
    		c.Do("") // 忽略所有返回值，但是全部flush进去
    	}
    	if true {
    		rand.Seed(time.Now().UnixNano())
    		c.Send("del", "set", "list", "zset")
    		c.Send("sadd", "set", "mm", "dd", "kk")
    		{
    			args := redis.Args{}.Add("zset")
    			args = args.Add(rand.Intn(100)).Add("mm")
    			args = args.Add(rand.Intn(100)).Add("dd")
    			args = args.Add(rand.Intn(100)).Add("kk")
    			c.Send("zadd", args...)
    		}
    		{
    			args := redis.Args{}.Add("zset")
    			args = args.Add(0).Add(-1).Add("withscores")
    			vals, err := redis.Values(c.Do("zrange", args...))
    			if err != nil {
    				panic(err)
    			}
    			var rets []struct {
    				Name  string
    				Score int
    			}
    			if err = redis.ScanSlice(vals, &rets); err != nil {
    				panic(err)
    			}
    			fmt.Println(rets)
    		}
    	}
    }

命令事务

  • trans.go
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    package main

    import (
    	"fmt"
    	"time"

    	"github.com/garyburd/redigo/redis"
    )

    func main() {
    	c, err := redis.Dial("tcp", fmt.Sprintf("%s:%d", "127.0.0.1", 6379), redis.DialPassword("123456"))
    	if err != nil {
    		panic(err)
    	}
    	defer (func() {
    		fmt.Println("connection close")
    		c.Close()
    	})()
    	c.Send("set", "score", 1000)
    	c.Do("watch", "score") // 只有do的时候会发生实际的通信
    	score, _ := redis.Int(c.Do("get", "score"))
    	scoreCpy := score
    	c.Send("multi") // 开启事务
    	time.Sleep(time.Second * 20)
    	c.Send("set", "score", score*2) // 此时我们另外开一个redis-client，set score 300
    	c.Do("exec")                    // 提交事务
    	score, _ = redis.Int(c.Do("get", "score"))
    	fmt.Println("score-old", scoreCpy, "\tscore-new", score)
    	// score-old 1000  score-new 300 发现没，因为我们的key中间被其他连接修改了，这个事务就被打断了，整体没有执行
    }

脚本事务

  • redis中加载了lua虚拟机，用来执行redis lua脚本；redis lua脚本的执行是原子性的；当某个脚本正在执行的时候，不会有其他命令或者脚本被执行。
  • double.lua
    1
    2
    3
    4
    5
    6
    7
    8
    local key = KEYS[1]
    local default = ARGV[1]
    if redis.call("exists", key) == 0 then
        redis.call("set", key, default)
    end
    local val = redis.call("get", key)
    redis.call("set", key, val * 2)
    return val*2
  • script.go
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    package main

    import (
    	"fmt"
    	"os"

    	"github.com/garyburd/redigo/redis"
    )

    func main() {
    	c, err := redis.Dial("tcp", fmt.Sprintf("%s:%d", "127.0.0.1", 6379), redis.DialPassword("123456"))
    	if err != nil {
    		panic(err)
    	}
    	defer (func() {
    		fmt.Println("connection close")
    		c.Close()
    	})()
    	var data []byte
    	data, err = os.ReadFile("./script/double.lua")
    	if err != nil {
    		fmt.Println("load double.lua error", err)
    		return
    	}
    	script := redis.NewScript(1, string(data)) // 脚本接收一个参数
    	script.Load(c)
    	if false { // 先确保key存在
    		c.Send("set", "score", 1000)
    		rpy, _ := redis.Int(script.Do(c, "score")) // 实际执行脚本，参数设置为score
    		fmt.Println(rpy)
    	}
    	if true { // 当key不存在时
    		rpy, _ := redis.Int(script.Do(c, "dd", 100))
    		fmt.Println(rpy)
    	}
    }

redigo订阅发布与stream

  • pubsub.go
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    package main

    import (
    	"fmt"

    	"github.com/garyburd/redigo/redis"
    )

    func main() {
    	sp, err := redis.Dial("tcp", fmt.Sprintf("%s:%d", "127.0.0.1", 6379), redis.DialPassword("123456"))
    	if err != nil {
    		panic(err)
    	}
    	defer (func() {
    		fmt.Println("connection close")
    		sp.Close()
    	})()
    	spc := redis.PubSubConn{Conn: sp}
    	if false { // 标准订阅
    		// Do == Send + Flush + Receive
    		// Subscribe == Send + Flush
    		// 所以我们后面直接receive就行了
    		spc.Subscribe("news.it")
    		for {
    			switch v := spc.Receive().(type) {
    			case redis.Message: // 重新起一个redis，PUBLISH news.it "apple new"，就会收到该消息
    				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
    			}
    		}
    	}
    	if true { // 模式订阅
    		spc.PSubscribe("new.*")
    		for {
    			switch v := spc.Receive().(type) {
    			case redis.PMessage:
    				// 重新起一个redis，PUBLISH new.show "apple2 new"，就会收到该消息
    				// PUBLISH new.ff "apple3 new"
    				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
    			}
    		}
    	}
    }

2024-12-13 该篇文章被 Cleofwine 归为分类: Go云原生

上一篇：【go分布式中间件】MySQL 下一篇：【go工程组件】分布式链路追踪OpenTelemetry(施工)
© 2023-2026 Cleofwine / Email / GitHub / 浙ICP备2024136127号
🌙赠我弯弯一枚月，也赠予我晚星⭐ 本站由Hexo驱动｜使用Hexo-theme-A4主题