直接用 chan 做任务队列易出阻塞、死锁、任务丢失等问题,因其仅为通信原语,缺乏重试、ACK、积压监控等生产级能力;应结合 select/default、sync.WaitGroup 或封装 TaskQueue,必要时换用 Redis/Kafka。

为什么直接用 chan 做任务队列容易出问题
Go 的 chan 天然适合生产者消费者模型,但直接裸用会导致阻塞、死锁或任务丢失。典型问题是:生产者往已满的无缓冲通道写入时永久阻塞;消费者 panic 后未关闭通道,导致其他 goroutine 无限等待;或者多个消费者竞争同一通道却没做任务确认机制,造成重复消费或漏消费。
关键点在于:chan 是通信原语,不是任务队列实现——它不提供重试、ACK、积压监控、优雅关闭等生产必需能力。
- 无缓冲通道(
make(chan int))要求收发双方同时就绪,否则阻塞 - 有缓冲通道(
make(chan int, 100))仅缓解压力,满后仍阻塞,且无法感知积压量 - 关闭通道后,接收方仍可读完剩余数据,但无法区分“空”和“已关闭”,易误判终止条件
用 chan + sync.WaitGroup 实现基础可靠模型
这是最轻量、可控性最强的实践方式,适用于中低频任务(如日志收集、配置同步),核心是让生产者不阻塞、消费者可退出、任务不丢失。
关键设计:
立即学习“go语言免费学习笔记(深入)”;
- 生产者用
select配合default实现非阻塞写入,失败时走本地重试或丢弃策略 - 消费者用
range遍历通道,配合sync.WaitGroup等待所有任务完成 - 关闭通道前确保所有生产者 goroutine 已退出,避免向已关闭通道写入 panic
func main() {
tasks := make(chan string, 10)
var wg sync.WaitGroup
// 启动 3 个消费者
for i := 0; i < 3; i++ {
wg.Add(1)
go func(id int) {
defer wg.Done()
for task := range tasks {
fmt.Printf("worker %d processing: %s\n", id, task)
time.Sleep(100 * time.Millisecond) // 模拟处理
}
}(i)
}
// 生产者:非阻塞写入
for i := 0; i < 20; i++ {
select {
case tasks <- fmt.Sprintf("task-%d", i):
default:
fmt.Printf("task-%d dropped: channel full\n", i)
}
}
close(tasks)
wg.Wait()}
加一层封装:带超时与错误反馈的 TaskQueue
当任务需要结果反馈、或必须保证至少一次交付时,裸通道就不够用了。此时应封装一个结构体,把输入通道、输出通道、错误通道、上下文控制统一管理。
注意点:
- 不要在消费者内部直接
panic,而是将错误发到errCh,由主逻辑统一处理 - 用
context.Context控制单个任务超时,避免某个慢任务拖垮整个队列 - 输出通道应为有缓冲(如
make(chan Result, len(tasks))),防止结果写入阻塞下一个任务处理
type Task struct {
ID string
Data []byte
Ctx context.Context
}
type Result struct {
TaskID string
Err error
Output interface{}
}
func NewTaskQueue(workers int, taskCh <-chan Task, resultCh chan<- Result) {
for i := 0; i < workers; i++ {
go func() {
for task := range taskCh {
select {
case <-task.Ctx.Done():
resultCh <- Result{TaskID: task.ID, Err: task.Ctx.Err()}
continue
default:
}
// 模拟处理
output := strings.ToUpper(string(task.Data))
resultCh <- Result{TaskID: task.ID, Output: output}
}
}()
}}
什么情况下该换用外部队列(Redis/Kafka)
当出现以下任一情况,说明已超出 chan 能力边界,硬撑只会增加维护成本:
- 需要跨进程/跨机器分发任务(
chan仅限单进程内) - 要求任务持久化,重启后不丢失(内存通道必然清空)
- 消费者处理时间波动大,需动态扩缩容(Go 程无法热增减,而 Kafka 可调分区数)
- 要查积压量、延迟、重试次数等指标(需额外埋点+存储,不如直接用 Redis 的
LLEN或 Kafka 的__consumer_offsets)
这时候别纠结“Go 就该用 channel”,真实系统里混用才是常态:Go 服务用 redis.Client 读取 LPUSH 的任务,处理完再 RPUSH 到结果队列——channel 只留在单机内部做 goroutine 协作,不越界。
最容易被忽略的是背压传递:哪怕用了 Redis,如果消费者拉取太快、处理太慢,还是会在本地堆积大量未处理 Task 结构体。所以无论底层用啥,select + default + 有限缓冲区这三板斧,在每一层都要存在。










