go - 如何确保 redis 订阅者在 Go (Golang) 中收到消息？

Question

我正在使用 gin 框架构建 API 服务器。一般来说，我正在构建 2 个项目。项目 'API' 和项目 'SOCKET'。 'API' 项目是将在 Android 中使用的主要 REST API，使用 gin 框架 (golang) 开发。 Project 'SOCKET' 是将使用 socket 连接的客户端的 socket 服务器，使用 node.js (Socket.IO)

过程是这样开始的:
用户A:作为请求者；连接到 “API”
用户B:作为响应者； B 连接到 "SOCKET"

用户 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” 中有一个订阅者将监听每一次发布，并将新消息发布到 channel 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 在线

我要的是:

1. 我们可以在 redis pub/sub 上创建类似 callback 的东西吗？如果订阅者由于离线或其他原因不接受消息，我们将关闭连接。这可能吗？
2. 在 Node.Js 中，我知道我可以使用超时功能，如果在特定时间没有收到消息，它将关闭订阅或发出任何事件，如何在 golang 上执行此操作？如果 User B 处于事件状态或离线状态，我需要通知 User A，以便他可以等待另一个时间来创建请求。
3. 如果什么都做不到，你对我这样做有什么建议？

我希望我的问题，可以理解，并且可以得到很好的回答。
*可能是一些代码，缺少变量。
** 我正在为 golang redis 使用这个库:go-redis

Answer 1

1) Redis 中没有回调。

2) 在 Go 中实现超时的常用方法是使用 channel 并选择 - 其中一个是您进行阻塞的 channel ，另一个 channel 在超时时接收消息。可以找到相关示例 here和 here for the docs

现在对于 (3)，您有一些方法选项。第一种是使用列表，从一侧推送(发布)并从另一侧弹出(订阅)。对于接收器，您可以使用 BLPOP 的 BRPOP - 分别阻止来自右侧或左侧的弹出。您可以将两者结合起来以获得持久的消息传递。

现在 PUBSUB 的一部分也取决于您要发布到的内容。如果您要发布到一个 channel ，该 channel 当且仅当有一个用户连接接收它(因此该 channel 只有一个订阅者)，您可以检查来自你的发布命令。它会告诉您它发布给了多少客户。如果 channel 仅由在线接收者订阅，您将返回“1”，如果用户离线则返回“0”。

第三个例子是将消息存储在一个有序集合中，时间戳作为分数。这将允许接收方连接并从上次连接时获取消息——但这假设在某个地方存在一些持久性——通常是客户端。您还需要对排序集进行一些清理事件。

在这种情况下需要考虑的其他一些事情是您最终是否使用复制，在这种情况下您必须明确考虑故障转移 - 尽管实际上在您描述的情况下您想要考虑断开连接和重新连接。 my post on reliable PUBSUB 中有这方面的具体示例.