17370845950

如何使用Golang构建基础消息队列模拟_Golang消息发送与消费实现方法
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,否则 panic

select 控制超时与非阻塞,避免 goroutine 卡死

真实业务中,你不能让生产者无限等待 channel 有空位,也不能让消费者空转轮询。必须用 selecttimeoutdefault 分支来兜底。

  • 发送端加超时:防止因消费者宕机或过慢导致整个流程 hang 住
  • 接收端加 default:实现“尽力而为”式消费,不阻塞主逻辑
  • 错误现象示例:fatal error: all goroutines are asleep - deadlock,基本就是没加 select 或没处理关闭逻辑
select {
case messages <- "新消息":
    fmt.Println("入队成功")
default:
    fmt.Println("队列已满,丢弃")
}

// 或带超时 select { case messages <- "新消息": 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,这些细节漏掉,换再重的组件也照样出错。