编辑“︁Gin消息队列集成”︁（章节）

== 实现方案 ==
本节以Redis Streams为例展示完整集成方案。

=== 准备工作 ===
安装所需库：
<syntaxhighlight lang="bash">
go get github.com/redis/go-redis/v9
</syntaxhighlight>

=== 生产者实现 ===
在路由处理函数中发布消息：

<syntaxhighlight lang="go">
package main

import (
	"github.com/gin-gonic/gin"
	"github.com/redis/go-redis/v9"
	"context"
)

func main() {
	r := gin.Default()
	rdb := redis.NewClient(&redis.Options{
		Addr: "localhost:6379",
	})

	r.POST("/tasks", func(c *gin.Context) {
		// 获取任务数据
		taskData := c.PostForm("data")
		
		// 发布到消息队列
		err := rdb.XAdd(context.Background(), &redis.XAddArgs{
			Stream: "gin_tasks",
			Values: map[string]interface{}{"data": taskData},
		}).Err()
		
		if err != nil {
			c.JSON(500, gin.H{"error": err.Error()})
			return
		}
		
		c.JSON(200, gin.H{"status": "queued"})
	})

	r.Run(":8080")
}
</syntaxhighlight>

=== 消费者实现 ===
独立消费者服务：

<syntaxhighlight lang="go">
package main

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

func main() {
	rdb := redis.NewClient(&redis.Options{
		Addr: "localhost:6379",
	})

	// 创建消费者组
	err := rdb.XGroupCreate(context.Background(), "gin_tasks", "gin_workers", "0").Err()
	if err != nil && err.Error() != "BUSYGROUP Consumer Group name already exists" {
		panic(err)
	}

	// 持续消费消息
	for {
		entries, err := rdb.XReadGroup(context.Background(), &redis.XReadGroupArgs{
			Group:    "gin_workers",
			Consumer: "worker1",
			Streams:  []string{"gin_tasks", ">"},
			Count:    1,
			Block:    0,
		}).Result()
		
		if err != nil {
			fmt.Println("Error reading:", err)
			continue
		}
		
		for _, msg := range entries[0].Messages {
			fmt.Printf("Processing task: %v\n", msg.Values["data"])
			// 处理业务逻辑...
			
			// 确认消息处理完成
			rdb.XAck(context.Background(), "gin_tasks", "gin_workers", msg.ID)
		}
	}
}
</syntaxhighlight>