手把手教你用 Go 写一个基于 Redis 的消息队列

消息队列，我们一般称它为 MQ（Message Queue），两个单词的结合，这两个英文单词想必大家都应该知道吧，其实最熟悉的还是 Queue 吧，即队列。队列是一种先进先出的数据结构，队列的使用还是比较普遍的。在现代软件开发中，消息队列是实现应用解耦、提高系统扩展性和健壮性的关键组件。Go 语言以其并发处理能力著称，是构建高性能消息队列的理想选择。本文将介绍一个使用 Go 语言和 Redis 实现的简单但功能完备的消息队列系统。

概述

消息队列系统通常由生产者（Producer）和消费者（Consumer）组成。生产者负责发送消息，而消费者则负责接收和处理这些消息。我们的消息队列系统使用 Redis 作为后端存储，利用其高性能和持久化特性来保证消息的可靠性。

程序结构

我们的系统由以下几个核心组件构成：

• Queue：队列管理器，负责创建队列、发布消息和启动消费者。
• Producer：消息生产者，用于向队列发送消息。
• Consumer：消息消费者，用于从队列接收并处理消息。
• Message：消息实体，定义了消息的结构和序列化方法。

核心代码解析

Queue 管理器

queue.go 文件定义了 Queue 结构和相关方法,它封装了一个用于消息队列处理的 redis.Client 实例，。Queue 结构也包含了 Redis 客户端、主题（topic）以及生产者和消费者实例。其中 NewQueue 函数用于创建一个新的消息队列实例，包括生产者和消费者。Start 方法用于启动消费者的监听，Publish 方法用于发布消息。完整代码如下：

package queue

import (
    "context"
    "github.com/go-redis/redis/v8"
    "sync"
)

const (
    HashSuffix = ":hash" // Redis 键后缀，用于哈希表
    SetSuffix = ":set" // Redis 键后缀，用于集合
)

var once sync.Once // 用于确保某些初始化代码只执行一次的全局变量

type Queue struct {
    ctx context.Context // 上下文

    // redis
    redis *redis.Client  // Redis客户端
    topic string         // 主题

    // producer and consumer
    producer *producer   // 生产者
    consumer *consumer   // 消费者

}

func NewQueue(ctx context.Context, redis *redis.Client, opts ...Option) *Queue {
    var queue *Queue

    // 使用sync.Once确保以下代码只执行一次
    once.Do(func() {
    	// 定义默认的选项
    	defaultOptions := Options{
    		topic:   "topic",
    		handler: defaultHander,  // 默认处理函数
    	}

    	// 应用传入的选项（opts）
    	for _, apply := range opts {
    		apply(&defaultOptions)
    	}

    	// 创建Queue实例
    	queue = &Queue{
    		ctx:      ctx,
    		redis:    redis,
    		topic:    defaultOptions.topic,
    		producer: NewProducer(ctx),  // 创建生产者
    		consumer: NewConsumer(ctx, defaultOptions.handler),  // 创建消费者，使用处理函数
    	}
    })

    // 返回Queue实例
    return queue

}

func (q *Queue) Start() {
    // 启动消费者的监听
    go q.consumer.listen(q.redis, q.topic)
}

func (q *Queue) Publish(msg *Message) (int64, error) {
    // 发布消息
    return q.producer.publish(q.redis, q.topic, msg)
}

• 你也发现了其中 NewQueue 函数接受一个上下文、Redis 客户端和一个可变数量的选项，用于创建一个新的队列实例。Option 定义了一个使用函数闭包实现的选项模式，用于生成具有不同选项的 Options 结构，其中包含了可配置的字段 topic 和 handler。WithTopic 和 WithHandler 是两个函数，它们接受相应的参数并返回对应的选项函数。当这些选项函数被应用到 Options 结构时，会设置结构体中相应的字段值。queue_option.go 代码如下：

package queue

type Option func(*Options)

type Options struct {
	topic   string
	handler handlerFunc
}

// WithTopic 用于设置选项中的 topic 字段
func WithTopic(topic string) Option {
	return func(opts *Options) {
		opts.topic = topic
	}
}

// WithHandler 用于设置选项中的 handler 字段
func WithHandler(handler handlerFunc) Option {
	return func(opts *Options) {
		opts.handler = handler
	}
}

消息实体

message.go 文件定义了 Message 结构，用于表示消息实体。它包含了消息的唯一标识符（ID）、创建时间、消费时间以及消息体（body）。NewMessage 函数用于创建消息实例。GetScore 函数返回消息的分数，GetId 函数返回消息的 ID，MarshalBinary 和 UnmarshalBinary 函数用于消息的序列化和反序列化。

package queue

import (
	"encoding/json"
	"github.com/satori/go.uuid"
	"time"
)

// Message 定义消息结构
type Message struct {
	Id          string      `json:"id"`
	CreateTime  time.Time   `json:"createTime"`
	ConsumeTime time.Time   `json:"consumeTime"`
	Body        interface{} `json:"body"`
}

// NewMessage 用于创建消息实体
func NewMessage(id string, consumeTime time.Time, body interface{}) *Message {
	if id == "" {
		id = uuid.NewV4().String()
	}
	return &Message{
		Id:          id,
		CreateTime:  time.Now(),
		ConsumeTime: consumeTime,
		Body:        body,
	}
}

// GetScore 用于返回消息的分数
func (m *Message) GetScore() float64 {
	return float64(m.ConsumeTime.Unix())
}

// GetId 用于返回消息的ID
func (m *Message) GetId() string {
	return m.Id
}

// MarshalBinary 用于将消息结构体序列化为二进制数据
func (m *Message) MarshalBinary() ([]byte, error) {
	return json.Marshal(m)
}

// UnmarshalBinary 用于将二进制数据反序列化为消息结构体
func (m *Message) UnmarshalBinary(data []byte) error {
	return json.Unmarshal(data, m)
}

生产者

producer.go 文件定义了 producer 结构和 publish 方法。生产者负责将消息发布到 Redis 的有序集合（sorted set）和哈希表（hash）中。代码创建了一个生产者类型并实现了发布消息到 Redis 的功能。在发布消息时，它首先将消息写入有序集合，然后将消息写入哈希表。这样，通过使用 NewProducer 函数创建生产者实例后，可以使用其 publish 方法将消息发布到 Redis 中保存。 producer.go 代码如下：

package queue

import (
	"context"
	"github.com/go-redis/redis/v8"
)

type producer struct {
	ctx context.Context
}

func NewProducer(ctx context.Context) *producer {
	return &producer{
		ctx: ctx,
	}
}

// Publish方法用于将消息发布到Redis中
func (p *producer) publish(redisClient *redis.Client, topic string, msg *Message) (int64, error) {
	z := &redis.Z{
		Score:  msg.GetScore(),
		Member: msg.GetId(),
	}

	// 将消息写入有序集合
	setKey := topic + SetSuffix
	n, err := redisClient.ZAdd(p.ctx, setKey, z).Result()
	if err != nil {
		return n, err
	}

	// 将消息写入哈希表
	hashKey := topic + HashSuffix
	return redisClient.HSet(p.ctx, hashKey, msg.GetId(), msg).Result()
}

消费者

consumer.go 文件定义了 consumer 结构和 listen 方法。消费者通过监听 Redis 的有序集合来获取消息，并从哈希表中检索消息详情，然后通过提供的处理器（handler）函数进行处理。建了一个消费者类型，并实现了监听消息队列中的消息并处理的功能。在 NewConsumer 方法中，创建了一个消费者实例并初始化了相关参数。在 listen 方法中，通过定时器定时从有序集合获取消息，并通过通道将数据送至处理函数进行最终逻辑处理。这样，通过使用 NewConsumer 函数创建消费者实例后，可以调用其 listen 方法实现对消息队列的监听和处理。consumer.go 代码如下：

package queue

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/go-redis/redis/v8"
	"log"
	"strconv"
	"time"
)

// handlerFunc 是用于处理消息的函数类型
type handlerFunc func(msg Message)

// 默认处理函数，打印消息
func defaultHandler(msg Message) {
	fmt.Println(msg)
}

// consumer 类型包含了消费者相关的字段和方法
type consumer struct {
	ctx      context.Context
	duration time.Duration
	ch       chan []string
	handler  handlerFunc
}

// NewConsumer 用于创建一个消费者实例
func NewConsumer(ctx context.Context, handler handlerFunc) *consumer {
	return &consumer{
		ctx:      ctx,
		duration: time.Second,
		ch:       make(chan []string, 1000),
		handler:  handler,
	}
}

// listen 方法用于监听消息队列中的消息并处理
func (c *consumer) listen(redisClient *redis.Client, topic string) {
	// 从哈希表中获取数据并处理
	go func() {
		for {
			select {
			case ret := <-c.ch:
				// 从哈希表中批量获取数据信息
				key := topic + HashSuffix
				result, err := redisClient.HMGet(c.ctx, key, ret...).Result()
				if err != nil {
					log.Println(err)
				}

				if len(result) > 0 {
					redisClient.HDel(c.ctx, key, ret...)
				}

				msg := Message{}
				for _, v := range result {
					// 由于哈希表和有序集合操作不是原子操作，可能会出现删除了集合中的数据但哈希表中数据未删除的情况
					if v == nil {
						continue
					}
					str := v.(string)
					json.Unmarshal([]byte(str), &msg)

					// 处理逻辑
					go c.handler(msg)
				}

			}
		}
	}()

	// 定时器用于定时获取消息并处理
	ticker := time.NewTicker(c.duration)
	defer ticker.Stop()
	for {
		select {
		case <-c.ctx.Done():  // 上下文取消，退出监听
			log.Println("consumer quit:", c.ctx.Err())
			return
		case <-ticker.C:  // 定时获取消息
			// 从 Redis 中读取数据
			min := strconv.Itoa(0)
			max := strconv.Itoa(int(time.Now().Unix()))
			opt := &redis.ZRangeBy{
				Min: min,
				Max: max,
			}

			key := topic + SetSuffix
			result, err := redisClient.ZRangeByScore(c.ctx, key, opt).Result()
			if err != nil {
				log.Fatal(err)
				return
			}

			// 获取到数据
			if len(result) > 0 {
				// 从有序集合中移除数据
				redisClient.ZRemRangeByScore(c.ctx, key, min, max)

				// 写入通道，进行哈希表处理
				c.ch <- result
			}
		}
	}
}

使用示例

假设我们有一个简单的场景，需要发送和处理用户行为日志。我们可以定义一个日志消息的生产者和消费者，main.go 如下所示：

var RedisQueue *queue.Queue
var Redis *redis.Client

func mian() {
	// 这里 redis 的链接自己写吧
	client := redis.NewClient(option)

	_, err := client.Ping(context.Background()).Result()

	if err != nil {
		log.Fatal(fmt.Sprintf("Connect to redis: %v", err))
	}

	Redis = client

	// 初始化 redis 消费队列
	InitRedisQueue()
}

// InitRedisQueue 创建延时队列
func InitRedisQueue() {
    // 创建队列
	RedisQueue = queue.NewQueue(context.Background(), Redis,
		queue.WithTopic("send-message"),
		queue.WithHandler(bot.CreateAndSendMessages))

    // 启动消费者
	RedisQueue.Start()
}

// 创建并发送消息
func CreateAndSendMessages() {
    id := uuid.NewV4().String()
    logMsg := queue.NewMessage(id, time.Now(), map[string]interface{}{"user_id": 123, "action": "login"})
    queue.Publish(logMsg)
    if err != nil {
		fmt.Println("send ", err)
	}
}

总结

本文介绍了一个基于 Go 语言和 Redis 实现的消息队列系统。该系统简单、高效，并且易于扩展。无论是在微服务架构中还是在需要解耦生产者和消费者的场景中，它都能提供强大的支持。请注意，这是一篇技术文章，您可以根据需要进行调整和补充。如果您有更多的细节或特定的要求，以便进一步用在生产环境中，请您完善更多的细节，比如：性能调优，特别是考虑到网络延迟和 Redis 实例的性能。

关注我获取更新