通过Redis实现一个异步请求-响应程序

在分布式系统中,经常需要不同的服务或组件之间进行通信和协作。传统的同步请求-响应模式虽然简单直观,但可能会导致阻塞和性能问题。为了解决这个问题,我们开发了一个基于 Redis 的异步请求-响应程序,实现了请求和响应的解耦,提高了系统的并发性和响应能力。

程序概述

该程序包含以下主要函数:

SyncRequest(ctx context.Context, id, funcKey string, params interface{}, timeout int) (interface{}, error)
- 用于发起请求,将请求信息存储在 Redis 中,并异步等待响应。
GetCallInfoById(ctx context.Context, id string) (string, interface{}, error)
- 根据请求 ID 从 Redis 中获取请求信息,供其他程序处理。
PublishReplyData(ctx context.Context, id string, data interface{}) error
- 将响应数据发布到 Redis 频道,供请求方接收。

通过这些函数,不同的程序可以实现异步的请求-响应模式,避免阻塞和提高并发性。

使用示例

假设我们有两个程序 A 和 B,需要进行异步通信。程序 A 发起请求,程序 B 处理请求并返回响应。

程序 A

package mainimport ("context""fmt""sagooiot/pkg/baseLogic"
)func main() {ctx := context.Background()id := "request_123" // 请求 IDfuncKey := "SomeFunction" // 函数键params := map[string]string{"key1": "value1", "key2": "value2"} // 请求参数// 发起异步请求response, err := baseLogic.SyncRequest(ctx, id, funcKey, params, 60)if err != nil {fmt.Println("Error:", err)return}// 处理响应fmt.Println("Response:", response)
}

在这个示例中,程序 A 调用 baseLogic.SyncRequest 发起一个异步请求,将请求信息存储在 Redis 中,并等待响应。超时时间设置为 60 秒。响应数据将被打印出来。

程序 B

package mainimport ("context""fmt""sagooiot/pkg/baseLogic"
)func main() {ctx := context.Background()id := "request_123" // 请求 ID// 从 Redis 获取请求信息funcKey, params, err := baseLogic.GetCallInfoById(ctx, id)if err != nil {fmt.Println("Error:", err)return}// 处理请求response := processRequest(funcKey, params)// 发布响应err = baseLogic.PublishReplyData(ctx, id, response)if err != nil {fmt.Println("Error:", err)return}
}func processRequest(funcKey string, params interface{}) interface{} {// 模拟处理请求的逻辑fmt.Println("Processing request:", funcKey, params)return "This is the response"
}