Go 用带缓冲 channel(如 make(chan string, 10))可构建轻量级内存消息队列,天然并发安全,适合开发调试等可丢消息场景;服务重启消息即丢失,缓冲大小需权衡内存与背压。
用 Go 构建基础消息队列,**不需要引入任何外部中间件,仅靠 channel 就能跑通生产-消费模型**;但它只适合轻量、非关键、可丢消息的场景(如开发调试、内部事件通知),一旦服务重启,所有未消费消息就彻底消失。
make(chan T, N) 创建带缓冲的内存队列这是最简可行路径:Go 原生 channel 天然支持并发安全、阻塞/非阻塞读写,加缓冲后就能充当“队列”——不是模拟,是直接可用的队列原语。
chan string 是同步通道,send 会卡住直到有 goroutine receive,不适合做队列make(chan string, 10) 才是真正意义上的“缓冲队列”,最多存 10 条,满则 send 阻塞(或用 select + default 实现非阻塞)make(chan string, 10000) 看似抗压,但内存占用陡增,且掩盖了消费者处理慢的真实问题messages := make(chan string, 10)
go func() {
for msg := range messages {
fmt.Println("处理:", msg)
time.Sleep(100 * time.Millisecond) // 模拟耗时操作
}
}()
messages <- "订单创建"
messages <- "用户登录"
close(messages) // 注意:关闭后不能再 send,否则 panicselect 控制超时与非阻塞,避免 goroutine 卡死真实业务中,你不能让生产者无限等待 channel 有空位,也不能让消费者空转轮询。必须用 select 加 timeout 或 default 分支来兜底。
default:实现“尽力而为”式消费,不阻塞主逻辑fatal error: all goroutines are asleep - deadlock,基本就是没加 select 或没处理关闭逻辑select {
case messages <- "新消息":
fmt.Println("入队成功")
default:
fmt.Println("队列已满,丢弃")
}
// 或带超时
select {
case mes
sages <- "新消息":
fmt.Println("入队成功")
case <-time.After(500 * time.Millisecond):
fmt.Println("超时,放弃发送")
}
直接裸用 chan 很快会遇到瓶颈:无法统计积压量、无法优雅关闭、无法加锁扩展持久化逻辑。此时应封装为结构体,把 chan 当作私有字段隐藏起来。
Send() 和 Receive() 方法,内部统一处理边界(满/空/关闭)sync.Mutex 不是为了保护 chan(它本身线程安全),而是为将来加计数器、日志、或切换为 Redis 后端预留钩子chan 字段本身——否则外部可绕过你的控制逻辑直接操作type SimpleQueue struct {
ch chan string
mu sync.Mutex
count int
}
func NewSimpleQueue(size int) *SimpleQueue {
return &SimpleQueue{
ch: make(chan string, size),
}
}
func (q *SimpleQueue) Send(msg string) bool {
q.mu.Lock()
defer q.mu.Unlock()
select {
case q.ch <- msg:
q.count++
return true
default:
return false
}
}
真正需要可靠、持久、可监控的消息队列时,channel 就该被替换成 Redis List 或 RabbitMQ;但所有这些高级方案,最初都始于一个带缓冲的 chan —— 它不是玩具,而是理解解耦本质的第一块砖。别急着上中间件,先搞懂为什么这里要加 buffer,为什么 close 后不能再 send,这些细节漏掉,换再重的组件也照样出错。