如何确保 redis 订阅者在 Go (Golang) 中收到消息?
How to ensure redis subscriber receive message in Go (Golang)?
我正在使用 gin 框架构建 API 服务器。一般来说,我正在构建 2 个项目。项目 'API'
和项目 'SOCKET'
。项目 'API'
是将在 Android 中使用的主要 REST API,使用 gin
框架 (golang) 开发。并且项目 'SOCKET'
是将使用套接字连接的客户端的套接字服务器,使用 node.js (Socket.IO
)
流程是这样开始的:
用户A:作为请求者;连接到 "API"
用户B:作为响应者; B 连接到 "SOCKET"
User A
从 android 调用 API requestData
,请求将由 "API"
的项目处理。而Project "API"
会记录请求,并发布到redis
作为 new_request
使用 pubsub
这是例如代码:
client := redis.NewClient(&redis.Options{
Addr: "localhost:6379",
Password: "", // no password set
DB: 0, // use default DB
})
pong, err := client.Ping().Result()
fmt.Println(pong, err)
if err !=nil {
fmt.Println("err",err);
}
pubsub, err := client.Subscribe("responseclient")
if err !=nil {
panic(err)
}
defer pubsub.Close()
err = client.Publish("new_request", "Example New Request").Err()
if err !=nil {
panic(err)
}
msg, err :=pubsub.ReceiveMessage()
if err != nil {
panic(err)
}
fmt.Println(msg.Channel, msg.Payload)
}
在项目 "SOCKET"
中有一个订阅者会监听每一次发布,并将新消息发布到频道 responseclient
这是示例代码:
ioApp.on ('connection' , function(socket) {
redisSub.on('new_request', function (channel, message) {
console.log(channel + ':' + message);
redisPub.publish("responseclient", JSON.stringify(res));
});
})
如果 用户 B 连接到 Socket.IO,此操作会很顺利。但是如果 用户 B 离线,或者没有连接到 socket.io,这将等待很长时间,直到我们手动终止或直到用户 B 在线
我要的是:
- 我们可以在 Redis pub/sub 上创建类似
callback
的东西吗?如果订阅者由于离线或其他原因不接受消息,我们将关闭连接。这可能吗?
- 在 Node.Js 我知道我可以使用超时功能,如果在特定时间没有收到消息,它将关闭订阅或发出任何事件,如何在 golang 上执行此操作?如果
User B
处于活动状态或离线状态,我需要通知 User A
,以便他可以等待另一个时间来创建请求。
- 如果没有办法,你对我这样做有什么建议?
希望我的问题,能看得懂,能得到很好的回答。
*可能有一些代码,缺少变量。
** 我正在为 golang redis 使用这个库:go-redis
1) Redis 中没有回调。
2) 在 Go 中实现超时的常用方法是使用通道和 select - 其中一个是您进行阻塞的通道,另一个通道在超时时接收消息。可以找到相关示例 here and here for the docs
现在对于 (3),您可以选择一些方法。第一种是使用列表,从一侧推送(发布)并从另一侧弹出(订阅)。对于您疯狂使用 BRPOP
或 BLPOP
的接收器 - 分别阻止来自右侧或左侧的弹出。您可以将两者结合起来以获得持久的消息传递。
现在 PUBSUB 的一部分也取决于您要发布到的内容。如果您发布到一个有订阅者的频道 当且仅当 有一个用户连接接收它(因此只有一个订阅者到该频道),您可以检查来自您的发布命令的响应。它会告诉您它发布给了多少客户。如果该频道仅由在线接收者订阅,您将返回“1”,如果用户离线则返回“0”。
第三个例子是将消息存储在一个有序集合中,时间戳作为分数。这将允许接收方连接并从上次连接时获取消息——但这假设在某个地方存在一些持久性——通常是客户端。您还需要对排序集进行一些清理 activity。
在这种情况下需要考虑的其他一些事情是您最终是否使用复制,在这种情况下您必须明确考虑故障转移 - 尽管实际上在您描述的情况下您想要考虑断开连接和重新连接。 my post on reliable PUBSUB.
中有这方面的具体示例
package main
import (
"context"
"fmt"
"time"
"github.com/go-redis/redis/v8"
)
var ctx = context.Background()
func main() {
rdb := redis.NewClient(&redis.Options{
Addr: "localhost:6379",
Password: "", // no password set
DB: 0, // use default DB
})
subscribe := rdb.Subscribe(ctx, "hello")
subscriptions := subscribe.ChannelWithSubscriptions(ctx, 1)
go func() {
var sentCount = 0
for {
rdb.Publish(ctx,"hello",time.Now().UnixNano())
sentCount++
if sentCount >300{
break
}
}
}()
for {
select {
case sub:=<-subscriptions:
fmt.Println(sub)
}
}
}
我正在使用 gin 框架构建 API 服务器。一般来说,我正在构建 2 个项目。项目 'API'
和项目 'SOCKET'
。项目 'API'
是将在 Android 中使用的主要 REST API,使用 gin
框架 (golang) 开发。并且项目 'SOCKET'
是将使用套接字连接的客户端的套接字服务器,使用 node.js (Socket.IO
)
流程是这样开始的:
用户A:作为请求者;连接到 "API"
用户B:作为响应者; B 连接到 "SOCKET"
User A
从 android 调用 API requestData
,请求将由 "API"
的项目处理。而Project "API"
会记录请求,并发布到redis
作为 new_request
使用 pubsub
这是例如代码:
client := redis.NewClient(&redis.Options{
Addr: "localhost:6379",
Password: "", // no password set
DB: 0, // use default DB
})
pong, err := client.Ping().Result()
fmt.Println(pong, err)
if err !=nil {
fmt.Println("err",err);
}
pubsub, err := client.Subscribe("responseclient")
if err !=nil {
panic(err)
}
defer pubsub.Close()
err = client.Publish("new_request", "Example New Request").Err()
if err !=nil {
panic(err)
}
msg, err :=pubsub.ReceiveMessage()
if err != nil {
panic(err)
}
fmt.Println(msg.Channel, msg.Payload)
}
在项目 "SOCKET"
中有一个订阅者会监听每一次发布,并将新消息发布到频道 responseclient
这是示例代码:
ioApp.on ('connection' , function(socket) {
redisSub.on('new_request', function (channel, message) {
console.log(channel + ':' + message);
redisPub.publish("responseclient", JSON.stringify(res));
});
})
如果 用户 B 连接到 Socket.IO,此操作会很顺利。但是如果 用户 B 离线,或者没有连接到 socket.io,这将等待很长时间,直到我们手动终止或直到用户 B 在线
我要的是:
- 我们可以在 Redis pub/sub 上创建类似
callback
的东西吗?如果订阅者由于离线或其他原因不接受消息,我们将关闭连接。这可能吗? - 在 Node.Js 我知道我可以使用超时功能,如果在特定时间没有收到消息,它将关闭订阅或发出任何事件,如何在 golang 上执行此操作?如果
User B
处于活动状态或离线状态,我需要通知User A
,以便他可以等待另一个时间来创建请求。 - 如果没有办法,你对我这样做有什么建议?
希望我的问题,能看得懂,能得到很好的回答。
*可能有一些代码,缺少变量。
** 我正在为 golang redis 使用这个库:go-redis
1) Redis 中没有回调。
2) 在 Go 中实现超时的常用方法是使用通道和 select - 其中一个是您进行阻塞的通道,另一个通道在超时时接收消息。可以找到相关示例 here and here for the docs
现在对于 (3),您可以选择一些方法。第一种是使用列表,从一侧推送(发布)并从另一侧弹出(订阅)。对于您疯狂使用 BRPOP
或 BLPOP
的接收器 - 分别阻止来自右侧或左侧的弹出。您可以将两者结合起来以获得持久的消息传递。
现在 PUBSUB 的一部分也取决于您要发布到的内容。如果您发布到一个有订阅者的频道 当且仅当 有一个用户连接接收它(因此只有一个订阅者到该频道),您可以检查来自您的发布命令的响应。它会告诉您它发布给了多少客户。如果该频道仅由在线接收者订阅,您将返回“1”,如果用户离线则返回“0”。
第三个例子是将消息存储在一个有序集合中,时间戳作为分数。这将允许接收方连接并从上次连接时获取消息——但这假设在某个地方存在一些持久性——通常是客户端。您还需要对排序集进行一些清理 activity。
在这种情况下需要考虑的其他一些事情是您最终是否使用复制,在这种情况下您必须明确考虑故障转移 - 尽管实际上在您描述的情况下您想要考虑断开连接和重新连接。 my post on reliable PUBSUB.
中有这方面的具体示例package main
import (
"context"
"fmt"
"time"
"github.com/go-redis/redis/v8"
)
var ctx = context.Background()
func main() {
rdb := redis.NewClient(&redis.Options{
Addr: "localhost:6379",
Password: "", // no password set
DB: 0, // use default DB
})
subscribe := rdb.Subscribe(ctx, "hello")
subscriptions := subscribe.ChannelWithSubscriptions(ctx, 1)
go func() {
var sentCount = 0
for {
rdb.Publish(ctx,"hello",time.Now().UnixNano())
sentCount++
if sentCount >300{
break
}
}
}()
for {
select {
case sub:=<-subscriptions:
fmt.Println(sub)
}
}
}