个人中心
gpt4 book ai didi

golang redigo发布订阅使用的方法

转载 作者：qq735679552 更新时间：2022-09-27 22:32:09 25 4
gpt4 key购买 nike

CFSDN坚持开源创造价值，我们致力于搭建一个资源共享平台，让每一个IT人在这里找到属于你的精彩世界.

这篇CFSDN的博客文章golang redigo发布订阅使用的方法由作者收集整理，如果你对这篇文章有兴趣，记得点赞哟.

redigo 对 发布订阅的使用

redigo 对redis 的发布订阅机制放在pubsub.go 中 订阅主题后 通过Receive() 函数接受发布订阅主题的消息 。

?
1
2
3
4
5
6
// Receive returns a pushed message as a Subscription, Message, Pong or error.
// The return value is intended to be used directly in a type switch as
// illustrated in the PubSubConn example.
func (c PubSubConn) Receive() interface{} {
     return c.receiveInternal(c.Conn.Receive())
}

返回的是一个空类型的interface{} , 由于空接口没有方法， 因此所有的类型都实现了空接口， 也就是说可以返回任意类型.

具体返回的类型 在receiveInternal（） 方法里面可以看到 。

?
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
func (c PubSubConn) receiveInternal(replyArg interface{}, errArg error) interface{} {
     reply, err := Values(replyArg, errArg)
     if err != nil {
         return err
     }
 
     var kind string
     reply, err = Scan(reply, &kind)
     if err != nil {
         return err
     }
 
     switch kind {
     case "message":
         var m Message
         if _, err := Scan(reply, &m.Channel, &m.Data); err != nil {
             return err
         }
         return m
     case "pmessage":
         var m Message
         if _, err := Scan(reply, &m.Pattern, &m.Channel, &m.Data); err != nil {
             return err
         }
         return m
     case "subscribe", "psubscribe", "unsubscribe", "punsubscribe":
         s := Subscription{Kind: kind}
         if _, err := Scan(reply, &s.Channel, &s.Count); err != nil {
             return err
         }
         return s
     case "pong":
         var p Pong
         if _, err := Scan(reply, &p.Data); err != nil {
             return err
         }
         return p
     }
     return errors.New("redigo: unknown pubsub notification")
}

目前返回 Message Subscription Pone 。

订阅的主题

收到消息之后通过注册的回调函数处理的方式， 所以代码中多了长map存放回调函数 。

?
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
package main
 
import (
     //"github.com/go-redis/redis"
     "fmt"
     "github.com/labstack/gommon/log"
     "github.com/gomodule/redigo/redis"
     "time"
 
     //"reflect"
     "unsafe"
)
 
type SubscribeCallback func (channel, message string)
 
type Subscriber struct {
     client redis.PubSubConn
     cbMap map[string]SubscribeCallback
}
 
func (c *Subscriber) Connect(ip string, port uint16) {
     conn, err := redis.Dial("tcp", "127.0.0.1:6379")
     if err != nil {
         log.Error("redis dial failed.")
     }
 
     c.client = redis.PubSubConn{conn}
     c.cbMap = make(map[string]SubscribeCallback)
 
     go func() {
         for {
             log.Info("wait...")
             switch res := c.client.Receive().(type) {
             case redis.Message:
                 channel := (*string)(unsafe.Pointer(&res.Channel))
                 message := (*string)(unsafe.Pointer(&res.Data))
                 c.cbMap[*channel](*channel, *message)
             case redis.Subscription:
                 fmt.Printf("%s: %s %d\n", res.Channel, res.Kind, res.Count)
             case error:
                 log.Error("error handle...")
                 continue
             }
         }
     }()
 
}
 
func (c *Subscriber) Close() {
     err := c.client.Close()
     if err != nil{
         log.Error("redis close error.")
     }
}
 
func (c *Subscriber) Subscribe(channel interface{}, cb SubscribeCallback) {
     err := c.client.Subscribe(channel)
     if err != nil{
         log.Error("redis Subscribe error.")
     }
 
     c.cbMap[channel.(string)] = cb
}
 
func TestCallback1(chann, msg string){
     log.Info("TestCallback1 channel : ", chann, " message : ", msg)
}
 
func TestCallback2(chann, msg string){
     log.Info("TestCallback2 channel : ", chann, " message : ", msg)
}
 
func TestCallback3(chann, msg string){
     log.Info("TestCallback3 channel : ", chann, " message : ", msg)
}
 
func main() {
 
     log.Info("===========main start============")
 
     var sub Subscriber
     sub.Connect("127.0.0.1", 6397)
     sub.Subscribe("test_chan1", TestCallback1)
     sub.Subscribe("test_chan2", TestCallback2)
     sub.Subscribe("test_chan3", TestCallback3)
 
     for{ // 这段代码的作用就是 阻止线程结束
         time.Sleep(1 * time.Second)
     }
}

运行main 文件 然后 看到 。

golang redigo发布订阅使用的方法

在redis 客户端 执行 发布信息 。

golang redigo发布订阅使用的方法

在控制台 看到监控结果 回调函数 执行的结果 。

golang redigo发布订阅使用的方法

发布

上面是订阅的代码和 代码要处理 的回调函数 发布直接使用默认的 Conn来Send Publish 就可以 redigo 的管道使用方法设计到三个函数 Do 函数也是下面这个函数的合并 。

  • c.Send()
  • c.Flush()
  • c.Receive()

解释： send() 方法吧命令写到缓冲区， flush() 把缓冲区的命令刷新到redis 服务器 receive() 函数接受redis 给予的 回应， 三个操作共同完成一套命令流程.

代码 。

?
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(
   //"github.com/go-redis/redis"
   "github.com/gomodule/redigo/redis"
   log "github.com/astaxie/beego/logs"
)
 
func main() {
 
   client, err := redis.Dial("tcp", "127.0.0.1:6379")
   if err != nil {
       log.Critical("redis dial failed.")
   }
   defer client.Close()
 
   _, err = client.Do("Publish", "test_chan1", "hello")
   if err != nil {
     log.Critical("redis Publish failed.")
   }
 
   _, err = client.Do("Publish", "test_chan2", "hello")
   if err != nil {
     log.Critical("redis Publish failed.")
   }
 
   _, err = client.Do("Publish", "test_chan3", "hello")
   if err != nil {
     log.Critical("redis Publish failed.")
   }
}

到此这篇关于golang redigo发布订阅使用的方法的文章就介绍到这了,更多相关golang redigo发布订阅内容请搜索我以前的文章或继续浏览下面的相关文章希望大家以后多多支持我！ 。

原文链接：https://blog.csdn.net/yangxiaodong88/article/details/96145648 。

最后此篇关于golang redigo发布订阅使用的方法的文章就讲到这里了,如果你想了解更多关于golang redigo发布订阅使用的方法的内容请搜索CFSDN的文章或继续浏览相关文章，希望大家以后支持我的博客！ 。

25 4 0
文章推荐： linux 安装配置lamp v2
文章推荐： springboot学习之构建简单项目搭建步骤详解
文章推荐： httpd.conf 二级域名设置方法
文章推荐： 定位并修复 Go 中的内存泄露问题
qq735679552
个人简介

我是一名优秀的程序员,十分优秀！

滴滴打车优惠券免费领取
滴滴打车优惠券
全站热门文章
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com