Go借助goroutine优化Redis回调任务 实现非阻塞高效处理

实现效果评估 #

你的调整是正确的，已经可以实现你需要的非阻塞高效处理效果：
原逻辑中数据拉取和业务处理是串行执行的，sendCallback的HTTP请求耗时会阻塞下一条数据的拉取。调整后你将单条数据的业务逻辑放到独立goroutine中异步执行，主线程拉取到数据后立刻回到循环拉取下一条，完全不会被业务逻辑的耗时卡住，只要Redis队列中有数据就会持续拉取启动处理，完全符合你的预期。

潜在问题与优化建议 #

当前实现可以跑通，但直接用于生产环境还存在几个风险点，建议优化：

• 无并发数限制：如果Redis队列瞬间积压大量任务，会启动数量不受控的goroutine，导致内存、CPU占用暴涨，同时过高的HTTP并发会打垮下游服务，或者耗尽本地端口、连接资源，建议用固定大小的Worker池控制并发上限。
• 未捕获panic：当前代码中JSON解析失败、HTTP请求出错都会直接触发panic，goroutine抛出的panic没有被捕获的话会导致整个进程退出，生产环境需要在业务goroutine中增加recover逻辑，捕获异常后记录日志即可，避免整个服务崩溃。
• 轮询拉取效率低：当前用RPop+2秒休眠的轮询方式，空队列下会有最多2秒的处理延迟，还会空耗CPU，建议换成Redis的BRPop命令，空队列时会阻塞等待新消息，有消息立刻返回，延迟更低也更省资源。
• 默认HTTP客户端性能不足：直接调用http.Post使用的是全局默认客户端，默认连接池参数不合理，高并发下容易出现连接复用率低、TIME_WAIT连接过多的问题，建议自定义初始化http.Client配置连接池参数。

优化后参考代码 #

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/go-redis/redis/v8"
	"net/http"
	"time"
)

var ctx = context.Background()
var httpClient *http.Client

// 初始化自定义HTTP客户端，优化连接池参数
func init() {
	transport := &http.Transport{
		MaxIdleConns:        100,
		MaxIdleConnsPerHost: 100,
		IdleConnTimeout:     30 * time.Second,
	}
	httpClient = &http.Client{
		Transport: transport,
		Timeout:   10 * time.Second, // 加请求超时，避免请求卡住
	}
}

func sendCallback(status string, sender string, recipient string, gatewayMessageId string) *http.Response {
	// 这里注意替换成你自己的postURL和responseBody构造逻辑
	postURL := "your callback url"
	var responseBody []byte // 你自己的请求体构造逻辑
	response, err := httpClient.Post(postURL, "application/json", http.NoBody)
	if err != nil {
		panic(err)
	}
	defer response.Body.Close() // 记得关闭响应体，避免连接泄漏
	fmt.Printf("POST : %s\n", status)
	return response
}

func callback(data string) {
	// 捕获当前goroutine的panic，避免进程崩溃
	defer func() {
		if r := recover(); r != nil {
			fmt.Printf("处理回调任务失败, 错误: %v, 数据: %s\n", r, data)
		}
	}()

	parsedData := make(map[string]interface{})
	err := json.Unmarshal([]byte(data), &parsedData)
	if err != nil {
		panic(err)
	}

	sender := parsedData["sender"].(string)
	recipient := parsedData["recipient"].(string)
	gatewayMessageId := parsedData["gateway_message_id"].(string)

	sendCallback("sent", sender, recipient, gatewayMessageId)
	sendCallback("delivered", sender, recipient, gatewayMessageId)
}

func main() {
	rdb := redis.NewClient(&redis.Options{
		Addr:     "127.0.0.1:6379",
		Password: "",
		DB:       0,
	})

	// 最大并发数，根据你的服务器性能和下游服务承受能力调整
	const maxWorker = 50
	// 带缓冲的任务通道，做削峰
	taskChan := make(chan string, 1000)

	// 启动固定数量的worker
	for i := 0; i < maxWorker; i++ {
		go func() {
			for data := range taskChan {
				callback(data)
			}
		}()
	}

	// 主线程负责拉取Redis数据丢入任务通道
	for {
		// BRPop阻塞等待最多5秒，有数据立刻返回
		res, err := rdb.BRPop(ctx, 5*time.Second, "callback").Result()
		if err != nil {
			if err == redis.Nil {
				// 超时没有数据，继续循环
				continue
			}
			fmt.Printf("拉取Redis队列失败: %v\n", err)
			time.Sleep(1 * time.Second)
			continue
		}
		// res[0]是队列名，res[1]是数据
		taskChan <- res[1]
	}
}

内容的提问来源于stack exchange，提问作者Ayodele Ashad