快速掌握 GO 之 RabbitMQ 结合 gin+gorm 案例

更多个人笔记见：
github个人笔记仓库
gitee 个人笔记仓库
个人学习，学习过程中还会不断补充～ （后续会更新在github和 gitee上）

gin+gorm框架例子

post-platform/
├── main.go           # Gin 服务（生产者）
├── rabbitmq.go       # RabbitMQ 操作
├── models/
│   └── post.go       # 帖子模型
├── db/
│   └── db.go         # 数据库连接和操作
├── consumer/
│   └── main.go       # 消费者（存储到 MySQL）
├── go.mod
└── go.sum

服务端生产者

• 定义 post.go

package models

type Post struct {
    Title   string `json:"title"`
    Content string `json:"content"`
}

gin 框架："go get github.com/gin-gonic/gin"

• main.go:

package main

import (
	"encoding/json"
	"log"
	"net/http"

	"github.com/gin-gonic/gin"
	"github.com/streadway/amqp"
)

func failOnError(err error, msg string) {
	if err != nil {
		log.Fatalf("%s: %s", msg, err)
	}
}

func main() {
	// 连接 RabbitMQ
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	failOnError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	defer ch.Close()

	q, err := ch.QueueDeclare("post_queue", false, false, false, false, nil)
	failOnError(err, "Failed to declare a queue")

	// 初始化 Gin
	r := gin.Default()

	// 提交帖子接口
	r.POST("/posts", func(c *gin.Context) {
		var post struct {
			Title   string `json:"title" binding:"required"`
			Content string `json:"content" binding:"required"`
		}

		if err := c.ShouldBindJSON(&post); err != nil {
			c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
			return
		}

		// 序列化帖子为 JSON
		postData, err := json.Marshal(post)
		if err != nil {
			c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to serialize post"})
			return
		}

		// 发送到 RabbitMQ
		err = ch.Publish(
			"",     // 交换机
			q.Name, // 队列名称
			false,  // 强制
			false,  // 立即
			amqp.Publishing{
				ContentType: "application/json",
				Body:        postData,
			})
		if err != nil {
			c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to publish to RabbitMQ"})
			return
		}

		c.JSON(http.StatusOK, gin.H{"message": "Post submitted successfully"})
	})

	r.Run(":8081")
}

数据库存储

gorm 框架，需要 go get：

    "gorm.io/driver/mysql"
    "gorm.io/gorm"

• db.go

package db

import (
	"log"
	"test/model"
	"gorm.io/driver/mysql"
	"gorm.io/gorm"
)

func InitDB() *gorm.DB {
	dsn := "root:password@tcp(localhost:3306)/dbname?charset=utf8mb4&parseTime=True&loc=Local"
	//根据情况填写password 和 dbname（具体的数据库和密码），这里用的本地 sql
	db, err := gorm.Open(mysql.Open(dsn), &gorm.Config{})
	if err != nil {
		log.Fatalf("Failed to connect to database: %v", err)
	}

	// 自动迁移，创建 posts 表
	err = db.AutoMigrate(&model.Post{})
	if err != nil {
		log.Fatalf("Failed to migrate database: %v", err)
	}

	return db
}

客户端消费者

• consumer.go

package main

import (
	"encoding/json"
	"log"
	"test/db"

	"github.com/streadway/amqp"
)

func failOnError(err error, msg string) {
	if err != nil {
		log.Fatalf("%s: %s", msg, err)
	}
}

type Post struct {
	Title   string `json:"title"`
	Content string `json:"content"`
}

func main() {
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	failOnError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	defer ch.Close()

	q, err := ch.QueueDeclare("post_queue", false, false, false, false, nil)
	failOnError(err, "Failed to declare a queue")

	// 初始化数据库
	db := db.InitDB()

	//消费消息
	msgs, err := ch.Consume(q.Name, "", true, false, false, false, nil)
	failOnError(err, "Failed to register a consumer")

	forever := make(chan bool)
	go func() {
		for d := range msgs {
			var posts Post
			if err := json.Unmarshal(d.Body, &posts); err != nil {
				log.Printf("Failed to unmarshal post: %v", err)
				continue
			}

			// 存储到数据库
			if err := db.Create(&posts).Error; err != nil {
				log.Printf("Failed to save post to database: %v", err)
				continue
			}
			log.Printf("Received post: Title=%s, Content=%s", posts.Title, posts.Content)
			// TODO: 存储到数据库（如 MySQL）
		}
	}()
	log.Printf(" [*] Waiting for posts. To exit press CTRL+C")
	<-forever // 等待程序退出,防止主线程退出，主动阻塞
}

gorm 中的 Create 是只要结构体的名字一样就会找对应的表，所以结构体命名为 Post/Posts都可以，虽然和 model 中的不一样，但是如果名字不一样，Create 函数就“找不到”

访问测试

分别终端运行程序后：
地址：http://localhost:8081/posts
发送内容：

{
    "title": "My First Post",
    "content": "Hello, world!"
}

可以发现能正确送达，同时能存储到数据库中