在Golang中实现简单的消息队列，核心是利用其原生的channel特性和goroutine并发能力，无需引入复杂的第三方中间件，就能满足小型项目的异步任务处理、模块解耦等需求。这种方式实现成本低，适配Golang的并发模型，适合对消息队列功能要求不复杂的场景。
消息队列的核心需求
一个基础的消息队列需要具备几个核心能力，我们在Golang实现时需要覆盖这些点：
- 消息生产：允许业务模块向队列中投递消息
- 消息消费：允许消费者从队列中获取消息并处理
- 异步处理：生产和消费过程不互相阻塞
- 基本的容错能力：比如消费者异常时不影响队列整体运行
基于channel的基础实现
Golang的channel天生支持线程安全的数据传递，非常适合作为消息队列的底层存储载体，结合goroutine可以实现异步的生产消费逻辑。
定义消息结构
首先我们需要定义消息的基础结构，方便扩展消息的属性和处理逻辑：
// 消息结构定义
type Message struct {
ID string // 消息唯一标识
Content interface{} // 消息内容
Retry int // 重试次数
}
实现队列核心结构
接下来定义队列的主体结构，包含消息通道、消费者管理、运行状态等字段：
package main
import (
"fmt"
"sync"
"time"
)
// 消息队列结构
type SimpleQueue struct {
msgChan chan Message // 消息通道
handlers []func(Message) // 消息处理函数列表
running bool // 队列运行状态
mu sync.RWMutex // 并发锁
wg sync.WaitGroup // 等待组，用于优雅关闭
}
初始化队列
提供初始化方法，设置通道缓冲大小，初始化队列状态：
// 初始化队列，bufferSize为通道缓冲大小
func NewSimpleQueue(bufferSize int) *SimpleQueue {
return &SimpleQueue{
msgChan: make(chan Message, bufferSize),
handlers: make([]func(Message), 0),
running: false,
}
}
注册消息处理器
允许外部注册消息处理函数，消费者获取到消息后会调用这些处理器处理消息：
// 注册消息处理函数
func (q *SimpleQueue) RegisterHandler(handler func(Message)) {
q.mu.Lock()
defer q.mu.Unlock()
q.handlers = append(q.handlers, handler)
}
启动队列
启动队列时会启动消费者协程，监听通道中的消息并处理：
// 启动队列
func (q *SimpleQueue) Start() {
q.mu.Lock()
if q.running {
q.mu.Unlock()
return
}
q.running = true
q.mu.Unlock()
// 启动消费者协程
q.wg.Add(1)
go q.consume()
fmt.Println("消息队列启动成功")
}
消费逻辑实现
消费者协程从通道中读取消息，遍历注册的处理器执行处理逻辑，同时捕获异常避免单个消息处理失败导致协程退出：
// 消费消息逻辑
func (q *SimpleQueue) consume() {
defer q.wg.Done()
for msg := range q.msgChan {
// 遍历所有处理器处理消息
q.mu.RLock()
handlers := q.handlers
q.mu.RUnlock()
for _, handler := range handlers {
func(h func(Message), m Message) {
defer func() {
if err := recover(); err != nil {
fmt.Printf("消息处理失败，消息ID：%s，错误：%vn", m.ID, err)
}
}()
h(m)
}(handler, msg)
}
}
}
生产消息方法
提供生产消息的接口，支持向队列中投递消息：
// 生产消息
func (q *SimpleQueue) Produce(msg Message) error {
q.mu.RLock()
if !q.running {
q.mu.RUnlock()
return fmt.Errorf("队列未启动")
}
q.mu.RUnlock()
// 向通道发送消息，通道满时会阻塞
q.msgChan <- msg
return nil
}
关闭队列
提供优雅关闭队列的方法，关闭通道并等待消费者处理完剩余消息：
// 关闭队列
func (q *SimpleQueue) Stop() {
q.mu.Lock()
if !q.running {
q.mu.Unlock()
return
}
q.running = false
q.mu.Unlock()
// 关闭通道，停止接收新消息
close(q.msgChan)
// 等待消费者协程退出
q.wg.Wait()
fmt.Println("消息队列已关闭")
}
完整使用示例
下面是一个完整的使用示例，演示如何初始化队列、注册处理器、生产消息和关闭队列：
func main() {
// 初始化队列，缓冲大小为100
queue := NewSimpleQueue(100)
// 注册消息处理器
queue.RegisterHandler(func(msg Message) {
fmt.Printf("处理消息，ID：%s，内容：%vn", msg.ID, msg.Content)
// 模拟处理耗时
time.Sleep(100 * time.Millisecond)
})
// 启动队列
queue.Start()
// 生产10条消息
for i := 0; i < 10; i++ {
msg := Message{
ID: fmt.Sprintf("msg_%d", i),
Content: fmt.Sprintf("测试消息内容_%d", i),
Retry: 0,
}
err := queue.Produce(msg)
if err != nil {
fmt.Printf("生产消息失败：%vn", err)
}
}
// 等待消息处理完成
time.Sleep(2 * time.Second)
// 关闭队列
queue.Stop()
}
实现注意事项
这种简单实现适合轻量场景，使用时需要注意几个问题：
- 通道缓冲大小需要根据业务消息量合理设置，过小会导致生产阻塞，过大可能占用过多内存
- 当前实现没有消息持久化能力，程序重启后未处理的消息会丢失，如果需要持久化可以结合文件或者轻量数据库存储消息
- 没有实现消息确认机制，如果处理器执行到一半异常，消息不会被重新投递，需要的话可以扩展重试逻辑
- 如果消息处理逻辑比较耗时，可以启动多个消费者协程提升处理效率，只需要在Start方法中启动多个consume协程即可
扩展方向
如果需要在基础实现上增加更多能力，可以从几个方向扩展：
- 增加消息持久化：在Produce时把消息写入文件或者数据库，队列启动时加载未处理的消息到通道
- 增加消息确认：消费者处理完成后返回确认结果，未确认的消息定期重新投递
- 增加优先级：使用多个不同优先级的channel，消费者优先从高优先级channel读取消息
- 增加监控：统计队列中的消息数量、处理耗时等指标，方便排查问题