AI 助理
备案控制台
开发者社区 数据库 文章 正文

Go WebSocket + Redis 实现轻量级的订阅和实时消息推送

2023-06-21 463
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
云数据库 Tair（兼容Redis），内存型 2GB
推荐场景：
通过缓存加速数据库访问
Redis 开源版，标准版 2GB
推荐场景：
搭建游戏排行榜
云解析 DNS，旗舰版 1个月
简介： Go WebSocket + Redis 实现轻量级的订阅和实时消息推送

上一篇介绍了Golang中封装WebSocket功能，让WebSocket更简单好用和稳定。


这里借助Redis自身的订阅和发布机制和WebSocket结合，实现轻量级的订阅发布和消息推送。本来消息订阅和推送打算用mqtt实现，但是这样还得有一个MqttBroker代理服务器，或采用网上开源的实现，或使用go语言自己实现个mqtt的broker。这都不够轻量级，这里介绍下借助redis的轻量级的实现。


大致框图如下：



涉及实时性和性能相关的服务可以直接在OnMessage里让客户端同后台业务直接交互实现。


关于提高性能的思考，首先看如果是高并发的情况下，瓶颈可能会在哪。


内部的两个redis客户端，一个负责发布，订阅，一个负责接收。当消息量大的情况下未必受用。那么首先负责发布的客户端，可考虑用redis的连接池实现。


消息的发布和订阅，固定为两个事件，一个是OnPublish,一个是OnSubcribe。并定义相关的报文结构如下：





收到的Publish事件，发布消息到Redis: 


// 接收到发布消息事件
  c.On("Publish", func(msg string) {
    // 将消息打印到控制台
    fmt.Printf("%s received publish: %s\n", c.Context().ClientIP(), msg)
    pubMsg := websocket.PushMsg{ID: c.ID()}
    err := json.Unmarshal([]byte(msg), &pubMsg)
    if err != nil {
      log.Printf("解析json串错误,err=", err)
      return
    }
    if pubMsg.Type != "pub" {
      log.Println("pub msg type error")
      return
    }
    //发布消息到Redis
    websocket.Publish(pubMsg.Topic, pubMsg.Payload)
  })


收到的订阅事件，发布消息到Redis:


// 接收到订阅的事件
  c.On("Subscribe", func(msg string) {
    // 将消息打印到控制台，c .Context（）是iris的http上下文。
    fmt.Printf("%s received subscribe: %s\n", c.Context().ClientIP(), msg)
    subMsg := websocket.SubMsg{ID: c.ID()}
    err := json.Unmarshal([]byte(msg), &subMsg)
    if err != nil {
      log.Printf("解析json串错误,err=", err)
      return
    }
    if pubMsg.Type != "pub" {
      log.Println("pub msg type error")
      return
    }
    //订阅到Redis
    sub.Subscribe(subMsg.Topic, subMsg.ID)
  })


开启一个Redis客户端，负责收到的消息：


func (c *Subscriber) Init(ws *Server) {
  conn := RedisClient.Get()
  c.client = redis.PubSubConn{conn}
  c.Ws = ws
  go func() {
    for {
      log.Println("redis wait...")
      switch res := c.client.Receive().(type) {
      case redis.Message:
        fmt.Printf("receive:%#v\n", res)
        topic := res.Channel
        message := string(res.Data)
        fnSubReceived(c.cbMap, topic, message)
      case redis.Subscription:
        fmt.Printf("%s: %s %d\n", res.Channel, res.Kind, res.Count)
      case error:
        log.Println("error handle", res)
        if IsConnError(res) {
          conn, err := RedisClient.Dial()
          if err != nil {
            log.Printf("err=%s\n", err)
          }
          c.client = redis.PubSubConn{conn}
        }
        continue
      }
    }
  }()
}


附完整实现：


package websocket
import (
  "log"
)
//订阅的消息格式定义
type SubMsg struct {
  ID    string `json:"id"`   //请求ID
  Type  string `json:"type"` //订阅时固定为sub,取消订阅时固定为unsub
  Topic string `json:"topic"`
  Param string `json:"param"`
}
//平台推送消息定义
type PushMsg struct {
  ID      string `json:"id"`
  Type    string `json:"type"` //发布，类型为pub
  Topic   string `json:"topic"`
  Payload string `json:"payload"`
  Result  string `json:"result"`
}
func Publish(topic string, msg string) (interface{}, error) {
  resp, err := Redo("Publish", topic, msg)
  if err != nil {
    log.Println(err)
  }
  return resp, err
}


package websocket
import (
  "errors"
  "fmt"
  "io"
  "log"
  "strings"
  "sync"
  "time"
  //"unsafe"
  "github.com/gomodule/redigo/redis"
)
var (
  // RD redis全局client
  RedisClient *redis.Pool
)
// InitRedis 初始设置
func InitRedis(host string, auth string, db int) error {
  // 连接Redis
  RedisClient = &redis.Pool{
    MaxIdle:     3,
    MaxActive:   4000,
    IdleTimeout: 180 * time.Second,
    Dial: func() (redis.Conn, error) {
      c, err := redis.Dial("tcp", host, redis.DialPassword(auth), redis.DialDatabase(db))
      if nil != err {
        return nil, err
      }
      return c, nil
    },
    TestOnBorrow: func(c redis.Conn, t time.Time) error {
      if time.Since(t) < time.Minute {
        return nil
      }
      _, err := c.Do("PING")
      return err
    },
  }
  rd := RedisClient.Get()
  defer rd.Close()
  c, err := redis.Dial("tcp", host, redis.DialPassword(auth), redis.DialDatabase(db))
  defer c.Close()
  if err != nil {
    fmt.Println("Connect to redis error", err)
    return err
  }
  fmt.Println("Connect to redis ok")
  return nil
}
func IsConnError(err error) bool {
  var needNewConn bool
  if err == nil {
    return false
  }
  if err == io.EOF {
    needNewConn = true
  }
  if strings.Contains(err.Error(), "use of closed network connection") {
    needNewConn = true
  }
  if strings.Contains(err.Error(), "connect: connection refused") {
    needNewConn = true
  }
  if strings.Contains(err.Error(), "connection closed") {
    needNewConn = true
  }
  return needNewConn
}
// 在pool加入TestOnBorrow方法来去除扫描坏连接
func Redo(command string, opt ...interface{}) (interface{}, error) {
  if RedisClient == nil {
    return "", errors.New("error,redis client is null")
  }
  rd := RedisClient.Get()
  defer rd.Close()
  var conn redis.Conn
  var err error
  var maxretry = 3
  var needNewConn bool
  resp, err := rd.Do(command, opt...)
  needNewConn = IsConnError(err)
  if needNewConn == false {
    return resp, err
  } else {
    conn, err = RedisClient.Dial()
  }
  for index := 0; index < maxretry; index++ {
    if conn == nil && index+1 > maxretry {
      return resp, err
    }
    if conn == nil {
      conn, err = RedisClient.Dial()
    }
    if err != nil {
      continue
    }
    resp, err := conn.Do(command, opt...)
    needNewConn = IsConnError(err)
    if needNewConn == false {
      return resp, err
    } else {
      conn, err = RedisClient.Dial()
    }
  }
  conn.Close()
  return "", errors.New("redis error")
}
type SubscribeCallback func(topicMap sync.Map, topic, msg string)
type Subscriber struct {
  client   redis.PubSubConn
  Ws       *Server //websocket
  cbMap    sync.Map
  CallBack interface {
    OnReceive(SubscribeCallback)
  }
}
var fnSubReceived SubscribeCallback
func (c *Subscriber) OnReceive(cb SubscribeCallback) {
  fnSubReceived = cb
}
func (c *Subscriber) Init(ws *Server) {
  conn := RedisClient.Get()
  c.client = redis.PubSubConn{conn}
  c.Ws = ws
  go func() {
    for {
      log.Println("redis wait...")
      switch res := c.client.Receive().(type) {
      case redis.Message:
        fmt.Printf("receive:%#v\n", res)
        topic := res.Channel
        message := string(res.Data)
        fnSubReceived(c.cbMap, topic, message)
      case redis.Subscription:
        fmt.Printf("%s: %s %d\n", res.Channel, res.Kind, res.Count)
      case error:
        log.Println("error handle", res)
        if IsConnError(res) {
          conn, err := RedisClient.Dial()
          if err != nil {
            log.Printf("err=%s\n", err)
          }
          c.client = redis.PubSubConn{conn}
        }
        continue
      }
    }
  }()
}
func (c *Subscriber) Close() {
  err := c.client.Close()
  if err != nil {
    log.Println("redis close error.")
  }
}
func (c *Subscriber) Subscribe(channel interface{}, clientid string) {
  err := c.client.Subscribe(channel)
  if err != nil {
    log.Println("redis Subscribe error.", err)
  }
  c.cbMap.Store(clientid, channel.(string))
}
func (c *Subscriber) PSubscribe(channel interface{}, clientid string) {
  err := c.client.PSubscribe(channel)
  if err != nil {
    log.Println("redis PSubscribe error.", err)
  }
  c.cbMap.Store(clientid, channel.(string))
}


package main
import (
  "fmt"
  "log"
  "sync"
  //"github.com/gin-contrib/cors"
  "encoding/json"
  "github.com/gin-gonic/gin"
  "net/http"
  "strings"
  "websockTest/websocket"
)
func main() {
  ws := websocket.New(websocket.Config{
    ReadBufferSize:  1024,
    WriteBufferSize: 1024,
  })
  ws.OnConnection(handleConnection)
  // 初始化连接redis
  var sub websocket.Subscriber
  err := websocket.InitRedis("127.0.0.1:6379", "", 0)
  if err != nil {
    fmt.Printf("InitRedis error: %s\n", err)
  } else {
    sub.Init(ws)
    //redis client收到的消息分发到websocket
    sub.OnReceive(func(topicMap sync.Map, topic, msg string) {
      fmt.Printf("sub msg received,topic=%s,msg=%s\n", topic, msg)
      topicMap.Range(func(k, v interface{}) bool {
        fmt.Println("topicMap:", k, v)
        if v.(string) == topic {
          conn := sub.Ws.GetConnection(k.(string))
          if conn != nil {
            conn.Write(1, []byte(msg))
          }
        }
        return true
      })
    })
  }
  r := gin.Default()
  //允许跨域
  //config := cors.DefaultConfig()
  //config.AllowOrigins = []string{"http://127.0.0.1:9090"}
  //r.Use(Cors())
  //静态资源
  r.Static("/static", "./static")
  r.LoadHTMLGlob("views/*")
  r.GET("/ws", ws.Handler())
  r.GET("/api/v3/device", ws.Handler())
  r.GET("/test", func(c *gin.Context) {
    c.HTML(http.StatusOK, "test.html", gin.H{
      "title": "this is a test",
    })
  })
  r.Run(":9090")
}
func handleConnection(c websocket.Connection) {
  fmt.Println("client connected,id=", c.ID())
  c.Write(1, []byte("welcome client"))
  // 从浏览器中读取事件
  c.On("chat", func(msg string) {
    // 将消息打印到控制台，c .Context（）是iris的http上下文。
    fmt.Printf("%s chat sent: %s\n", c.Context().ClientIP(), msg)
    // 将消息写回客户端消息所有者：
    // c.Emit("chat", msg)
    c.To(websocket.All).Emit("chat", msg)
  })
  c.OnMessage(func(msg []byte) {
    fmt.Println("received msg:", string(msg))
    c.Write(1, []byte("hello aa"))
    c.To(websocket.All).Emit("chat", msg)
  })
  c.OnDisconnect(func() {
    fmt.Println("client Disconnect,id=", c.ID())
  })
  // 接收到发布消息事件
  c.On("Publish", func(msg string) {
    // 将消息打印到控制台，c .Context（）是iris的http上下文。
    fmt.Printf("%s received publish: %s\n", c.Context().ClientIP(), msg)
    pubMsg := websocket.PushMsg{ID: c.ID()}
    err := json.Unmarshal([]byte(msg), &pubMsg)
    if err != nil {
      log.Printf("解析json串错误,err=", err)
      return
    }
    if pubMsg.Type != "pub" {
      log.Println("pub msg type error")
      return
    }
    //发布消息到Redis
    websocket.Publish(pubMsg.Topic, pubMsg.Payload)
  })
  // 接收到订阅的事件
  c.On("Subscribe", func(msg string) {
    // 将消息打印到控制台，c .Context（）是iris的http上下文。
    fmt.Printf("%s received subscribe: %s\n", c.Context().ClientIP(), msg)
    subMsg := websocket.SubMsg{ID: c.ID()}
    err := json.Unmarshal([]byte(msg), &subMsg)
    if err != nil {
      log.Printf("解析json串错误,err=", err)
      return
    }
    if pubMsg.Type != "pub" {
      log.Println("pub msg type error")
      return
    }
    //订阅到Redis
    sub.Subscribe(subMsg.Topic, subMsg.ID)
  })
}


文章标签：
云数据库 Tair（兼容 Redis）
云解析DNS
Go
NoSQL
Redis
关键词：
云数据库 Tair（兼容 Redis）订阅
WebSocket消息推送
Go redis
go云数据库 Tair（兼容 Redis）
Go websocket
相关实践学习
基于Redis实现在线游戏积分排行榜
本场景将介绍如何基于Redis数据库实现在线游戏中的游戏玩家积分排行榜功能。
云数据库 Redis 版使用教程
云数据库Redis版是兼容Redis协议标准的、提供持久化的内存数据库服务，基于高可靠双机热备架构及可无缝扩展的集群架构，满足高读写性能场景及容量需弹性变配的业务需求。 产品详情：https://www.aliyun.com/product/kvstore &nbsp; &nbsp; ------------------------------------------------------------------------- 阿里云数据库体验：数据库上云实战 开发者云会免费提供一台带自建MySQL的源数据库&nbsp;ECS 实例和一台目标数据库&nbsp;RDS实例。跟着指引，您可以一步步实现将ECS自建数据库迁移到目标数据库RDS。 点击下方链接，领取免费ECS&amp;RDS资源，30分钟完成数据库上云实战！https://developer.aliyun.com/adc/scenario/51eefbd1894e42f6bb9acacadd3f9121?spm=a2c6h.13788135.J_3257954370.9.4ba85f24utseFl
特立独行的猫a
目录
相关文章
赵渝强老师
|
7天前
|
消息中间件 NoSQL Redis
【赵渝强老师】Redis的消息发布与订阅
本文介绍了Redis实现消息队列的两种场景：发布者订阅者模式和生产者消费者模式。其中，发布者订阅者模式通过channel频道进行解耦，订阅者监听特定channel的消息，当发布者向该channel发送消息时，所有订阅者都能接收到消息。文章还提供了相关操作命令及示例代码，展示了如何使用Redis实现消息的发布与订阅。
赵渝强老师
15 0
武子康
|
1月前
|
SQL 分布式计算 NoSQL
大数据-42 Redis 功能扩展 发布/订阅模式 事务相关的内容 Redis弱事务
大数据-42 Redis 功能扩展 发布/订阅模式 事务相关的内容 Redis弱事务
武子康
24 2
何雨晨
|
1月前
|
负载均衡 网络协议 C#
C#实现WebSocket实时消息推送技术详解
C#实现WebSocket实时消息推送技术详解
何雨晨
37 1
游客j4mujezz7vm7y
|
2月前
|
NoSQL Go API
go语言操作Redis
go语言操作Redis
游客j4mujezz7vm7y
34 2
落雨便归尘
|
3月前
|
消息中间件 存储 NoSQL
redis实战——go-redis的使用与redis基础数据类型的使用场景(一)
本文档介绍了如何使用 Go 语言中的 `go-redis` 库操作 Redis 数据库
落雨便归尘
172 0
redis实战——go-redis的使用与redis基础数据类型的使用场景(一)
小王老师呀
|
2月前
|
消息中间件 NoSQL Go
PHP转Go系列 | ThinkPHP与Gin框架之Redis延时消息队列技术实践
【9月更文挑战第7天】在从 PHP 的 ThinkPHP 框架迁移到 Go 的 Gin 框架时，涉及 Redis 延时消息队列的技术实践主要包括：理解延时消息队列概念，其能在特定时间处理消息，适用于定时任务等场景；在 ThinkPHP 中使用 Redis 实现延时队列；在 Gin 中结合 Go 的 Redis 客户端库实现类似功能；Go 具有更高性能和简洁性，适合处理大量消息。迁移过程中需考虑业务需求及系统稳定性。
小王老师呀
57 0
wljslmz
|
3月前
|
存储 设计模式 安全
空结构体：Go 语言中的轻量级占位符
【8月更文挑战第31天】
wljslmz
41 0
wljslmz
|
3月前
|
Go 调度
Goroutine：Go语言的轻量级并发机制
【8月更文挑战第31天】
wljslmz
37 0
1603458909890196
|
3月前
|
NoSQL Go Redis
用 Go + Redis 实现分布式锁
用 Go + Redis 实现分布式锁
1603458909890196
42 0
1603458909890196
|
3月前
|
前端开发 Go 开发者
用 Go + WebSocket 快速实现一个 chat 服务
用 Go + WebSocket 快速实现一个 chat 服务
1603458909890196
59 0

热门文章

最新文章

  • 1
    java开发之使用websocket实现web客户端与服务器之间的实时通讯
  • 2
    Tomcat 7的WebSocket实现(下)
  • 3
    基于 WebSocket 实现 WebGL 3D 拓扑图实时数据通讯同步（二）
  • 4
    Netty实战，Springboot + netty +websocket 实现推送消息（附源码）
  • 5
    python 版websocket实现
  • 6
    WebSocket心跳机制（笔记大全）
  • 7
    解决浏览器不兼容websocket
  • 8
    VUE+websocket编写实现PC web端控制摄像头
  • 9
    websocket @Autowired注入为null问题的解决办法
  • 10
    SpringBoot轻松整合WebSocket，实现Web在线聊天室
  • 1
    为什么要学习Go语言？
    57
  • 2
    Go语言map并发安全，互斥锁和读写锁谁更优？
    57
  • 3
    Go语言map并发安全使用的正确姿势
    60
  • 4
    Go语言中的map为什么默认不是并发安全的？
    56
  • 5
    go语言中fmt格式化包和内置函数汇总
    74
  • 6
    go语言中格式化输出的占位符
    59
  • 7
    零值在go语言和初始化数据
    95
  • 8
    Go语言设计模式：使用Option模式简化类的初始化
    73
  • 9
    Go 语言性能优化技巧
    90
  • 10
    2019江行智能 后端开发工程师Go 面试题
    57

    • 相关课程

    更多
  • Go语言核心编程 - 数据结构和算法
  • Go语言核心编程 - 基础语法、数组、切片、Map
  • Go语言核心编程 - 面向对象、文件、单元测试、反射、TCP编程
  • Go语言完全自学手册图文教程
  • 云数据库 Redis 版使用教程
  • Redis数据库入门

    • 相关电子书

    更多
  • Redis在唯品会的应用实践——架构演进与功能定制
  • 微博的Redis定制之路
  • 云数据库Redis版的开源之路

    • 相关实验场景

    更多
  • 通过轻量消息队列（原MNS）主题HTTP订阅+ARMS实现自定义数据多渠道告警
  • 消息队列RocketMQ版：基础消息收发功能体验
  • 消息队列RocketMQ版：定时消息通知功能体验
  • 基于ROS的消息发布与订阅
  • 如何通过数据库Redis内网访问
  • RocketMQ中使用Java客户端发送消息和消费的应用
    • 下一篇
    阿里云OSS设置跨域访问