Go Channels阻塞问题：基于Go+Redis的消息队列自动扩缩容问询

嘿，我帮你把这个Go+Redis消息队列的自动扩缩容方案完善好了，核心就是根据消息积压量动态调整发送协程的数量，既保证吞吐量又不浪费资源。下面是完整的实现思路和代码：

核心设计思路 #

• 以消息通道的积压消息数作为扩缩容触发指标
• 设定maxSender（最大协程数）和minSender（最小协程数，建议设为1）避免极端情况
• 用senderRatio单协程处理阈值计算所需协程数：需要协程数 = 当前积压消息数 / senderRatio（向上取整）
• 定时检查通道状态，动态新增或优雅销毁协程

完整实现代码 #

package main

import (
	"context"
	"fmt"
	"sync"
	"time"

	"github.com/go-redis/redis/v8"
)

// Message 定义消息结构体，可根据业务需求扩展字段
type Message struct {
	Payload string
}

var (
	// 消息通道最大缓存量，避免内存溢出
	maxMessages = float64(100000)
	// 最大Redis发送协程数，限制资源占用
	maxSender = float64(5)
	// 最小发送协程数，保证基础吞吐量
	minSender = float64(1)
	// 单协程处理消息阈值：超过该数则触发扩容
	senderRatio = float64(20000)
	// 消息缓存通道
	messages = make(chan Message, int(maxMessages))
	// Redis客户端（go-redis本身协程安全，可多协程共用）
	rdb = redis.NewClient(&redis.Options{
		Addr:     "localhost:6379",
		Password: "", // 替换为你的Redis密码
		DB:       0,  // 使用默认数据库
	})
	// 当前活跃发送协程数
	currentSenders = 0
	// 协程数修改互斥锁，保证并发安全
	senderMutex = sync.Mutex{}
)

// sendToRedis 单个发送协程逻辑：从通道取消息并发送到Redis
func sendToRedis(ctx context.Context, wg *sync.WaitGroup, exitChan chan struct{}) {
	defer wg.Done()
	defer func() {
		senderMutex.Lock()
		currentSenders--
		senderMutex.Unlock()
		fmt.Printf("发送协程退出，当前协程数: %d\n", currentSenders)
	}()

	for {
		select {
		case <-ctx.Done():
			return
		case <-exitChan:
			// 处理完当前消息后退出（如果有的话）
			select {
			case msg, ok := <-messages:
				if ok {
					sendMsgToRedis(msg)
				}
			default:
			}
			return
		case msg, ok := <-messages:
			if !ok {
				return
			}
			sendMsgToRedis(msg)
		}
	}
}

// sendMsgToRedis 封装Redis发送逻辑，便于统一处理失败
func sendMsgToRedis(msg Message) {
	err := rdb.Publish(context.Background(), "message_queue", msg.Payload).Err()
	if err != nil {
		fmt.Printf("发送消息到Redis失败: %v\n", err)
		// 可选：将失败消息重新放入通道或存入死信队列
		// messages <- msg
	}
}

// scaleSenders 动态扩缩容协程的核心逻辑
func scaleSenders(ctx context.Context, wg *sync.WaitGroup, exitChans []chan struct{}) {
	ticker := time.NewTicker(2 * time.Second) // 每2秒检查一次，可根据业务调整
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			senderMutex.Lock()
			backlog := float64(len(messages))
			// 计算所需协程数，向上取整避免小数
			neededSenders := ceil(backlog / senderRatio)
			// 限制在最大最小范围内
			if neededSenders > maxSender {
				neededSenders = maxSender
			}
			if neededSenders < minSender {
				neededSenders = minSender
			}

			// 扩容逻辑：新增协程
			for currentSenders < int(neededSenders) {
				exitChan := make(chan struct{})
				exitChans = append(exitChans, exitChan)
				currentSenders++
				wg.Add(1)
				go sendToRedis(ctx, wg, exitChan)
				fmt.Printf("新增发送协程，当前协程数: %d\n", currentSenders)
			}

			// 缩容逻辑：向多余协程发送退出信号
			for currentSenders > int(neededSenders) {
				if len(exitChans) == 0 {
					break
				}
				exitChan := exitChans[0]
				exitChans = exitChans[1:]
				close(exitChan)
				currentSenders--
			}
			senderMutex.Unlock()
		}
	}
}

// ceil 自定义向上取整函数
func ceil(f float64) float64 {
	if f > float64(int(f)) {
		return float64(int(f) + 1)
	}
	return f
}

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	var wg sync.WaitGroup
	var exitChans []chan struct{}

	// 启动初始协程（最小协程数）
	senderMutex.Lock()
	currentSenders = int(minSender)
	for i := 0; i < int(minSender); i++ {
		exitChan := make(chan struct{})
		exitChans = append(exitChans, exitChan)
		wg.Add(1)
		go sendToRedis(ctx, wg, exitChan)
	}
	senderMutex.Unlock()
	fmt.Printf("初始启动协程数: %d\n", currentSenders)

	// 启动扩缩容监控协程
	wg.Add(1)
	go func() {
		defer wg.Done()
		scaleSenders(ctx, wg, exitChans)
	}()

	// 模拟消息生产逻辑，可替换为实际业务的消息来源
	go func() {
		for i := 0; i < 150000; i++ {
			messages <- Message{Payload: fmt.Sprintf("message_%d", i)}
			if i%20000 == 0 {
				time.Sleep(100 * time.Millisecond) // 模拟生产速度波动
			}
		}
		close(messages)
		fmt.Println("消息生产完成，关闭消息通道")
	}()

	// 等待所有协程执行完成
	wg.Wait()
	fmt.Println("所有消息处理完成")
}

关键细节说明 #

• 协程安全：用sync.Mutex保护currentSenders的修改，避免并发竞争问题
• 优雅缩容：给每个协程分配独立的退出通道，缩容时发送退出信号，让协程处理完当前消息后再退出，避免消息丢失
• 失败处理：Redis发送失败的消息可根据需求重新放入通道或存入死信队列，保证消息可靠性
• 参数调优：senderRatio、检查间隔、最大最小协程数需要根据Redis性能和业务吞吐量调整，比如Redis性能强可适当调大senderRatio

内容的提问来源于stack exchange，提问作者Sam White