From 4225b026d71f36820635ebcdefdaa480497bc7c3 Mon Sep 17 00:00:00 2001 From: Leon <147289645+LeoninCS@users.noreply.github.com> Date: Mon, 29 Dec 2025 04:13:13 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=B7=BB=E5=8A=A0rabbitMQ?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/cmd/main.go | 23 +++- backend/configs/config.yaml | 15 ++- backend/internal/config/loadconfig.go | 16 +++ .../internal/middleware/rabbitmq/rabbitMQ.go | 106 ++++++++++++++++++ .../internal/middleware/rabbitmq/socialMQ.go | 77 +++++++++++++ 5 files changed, 230 insertions(+), 7 deletions(-) create mode 100644 backend/internal/middleware/rabbitmq/rabbitMQ.go create mode 100644 backend/internal/middleware/rabbitmq/socialMQ.go diff --git a/backend/cmd/main.go b/backend/cmd/main.go index 0d4f9c1..0f8b205 100644 --- a/backend/cmd/main.go +++ b/backend/cmd/main.go @@ -5,6 +5,7 @@ import ( "feedsystem_video_go/internal/config" "feedsystem_video_go/internal/db" apphttp "feedsystem_video_go/internal/http" + rabbitmq "feedsystem_video_go/internal/middleware/rabbitmq" rediscache "feedsystem_video_go/internal/middleware/redis" "log" "strconv" @@ -12,14 +13,14 @@ import ( ) func main() { - // Load config + // 加载配置 log.Printf("Loading config from configs/config.yaml") cfg, err := config.Load("configs/config.yaml") if err != nil { log.Fatalf("Failed to load config: %v", err) } - // Connect database + // 连接数据库 //log.Printf("Database config: %v", cfg.Database) sqlDB, err := db.NewDB(cfg.Database) if err != nil { @@ -30,8 +31,8 @@ func main() { } defer db.CloseDB(sqlDB) - // Connect redis (optional, used for caching) - cache, err := rediscache.NewFromEnv() + // 连接 Redis (可选，用于缓存) + cache, err := rediscache.NewFromEnv(&cfg.Redis) if err != nil { log.Printf("Redis config error (cache disabled): %v", err) cache = nil @@ -48,8 +49,18 @@ func main() { } } - // Set router - r := apphttp.SetRouter(sqlDB, cache) + // 连接 RabbitMQ (可选，用于消息队列) + rmq, err := rabbitmq.NewRabbitMQ(&cfg.RabbitMQ) + if err != nil { + log.Printf("RabbitMQ config error (disabled): %v", err) + rmq = nil + } else { + defer rmq.Close() + log.Printf("RabbitMQ connected") + } + + // 设置路由 + r := apphttp.SetRouter(sqlDB, cache, rmq) log.Printf("Server is running on port %d", cfg.Server.Port) if err := r.Run(":" + strconv.Itoa(cfg.Server.Port)); err != nil { log.Fatalf("Failed to run server: %v", err) diff --git a/backend/configs/config.yaml b/backend/configs/config.yaml index 0358a87..223f2c5 100644 --- a/backend/configs/config.yaml +++ b/backend/configs/config.yaml @@ -6,4 +6,17 @@ database: port: 3306 user: root password: 123456 - dbname: feedsystem \ No newline at end of file + dbname: feedsystem + +redis: + host: localhost + port: 6379 + password: 123456 + db: 0 + +rabbitmq: + host: localhost + port: 5672 + username: admin + password: password123 + \ No newline at end of file diff --git a/backend/internal/config/loadconfig.go b/backend/internal/config/loadconfig.go index 763f11b..5a443e3 100644 --- a/backend/internal/config/loadconfig.go +++ b/backend/internal/config/loadconfig.go @@ -9,6 +9,8 @@ import ( type Config struct { Server ServerConfig `yaml:"server"` Database DatabaseConfig `yaml:"database"` + Redis RedisConfig `yaml:"redis"` + RabbitMQ RabbitMQConfig `yaml:"rabbitmq"` } type ServerConfig struct { @@ -23,6 +25,20 @@ type DatabaseConfig struct { DBName string `yaml:"dbname"` } +type RedisConfig struct { + Host string `yaml:"host"` + Port int `yaml:"port"` + Password string `yaml:"password"` + DB int `yaml:"db"` +} + +type RabbitMQConfig struct { + Host string `yaml:"host"` + Port int `yaml:"port"` + Username string `yaml:"username"` + Password string `yaml:"password"` +} + func Load(filename string) (Config, error) { data, err := ioutil.ReadFile(filename) if err != nil { diff --git a/backend/internal/middleware/rabbitmq/rabbitMQ.go b/backend/internal/middleware/rabbitmq/rabbitMQ.go new file mode 100644 index 0000000..108ee75 --- /dev/null +++ b/backend/internal/middleware/rabbitmq/rabbitMQ.go @@ -0,0 +1,106 @@ +package rabbitmq + +import ( + "context" + "encoding/json" + "errors" + "feedsystem_video_go/internal/config" + "strconv" + "time" + + amqp "github.com/rabbitmq/amqp091-go" +) + +type RabbitMQ struct { + conn *amqp.Connection + ch *amqp.Channel +} + +func NewRabbitMQ(cfg *config.RabbitMQConfig) (*RabbitMQ, error) { + if cfg == nil { + return nil, errors.New("rabbitmq config is nil") + } + url := "amqp://" + cfg.Username + ":" + cfg.Password + "@" + cfg.Host + ":" + strconv.Itoa(cfg.Port) + "/" + conn, err := amqp.Dial(url) + if err != nil { + return nil, err + } + ch, err := conn.Channel() + if err != nil { + return nil, err + } + return &RabbitMQ{conn: conn, ch: ch}, nil +} + +func (r *RabbitMQ) Close() error { + if r == nil || r.ch == nil || r.conn == nil { + return nil + } + if err := r.ch.Close(); err != nil { + return err + } + if err := r.conn.Close(); err != nil { + return err + } + return nil +} + +func (r *RabbitMQ) DeclareTopic(exchange string, queue string, bindingKey string) error { + if r == nil || r.ch == nil { + return errors.New("rabbitmq is not initialized") + } + if exchange == "" || queue == "" || bindingKey == "" { + return errors.New("exchange/queue/bindingKey is required") + } + + if err := r.ch.ExchangeDeclare( + exchange, + "topic", + true, + false, + false, + false, + nil, + ); err != nil { + return err + } + + q, err := r.ch.QueueDeclare( + queue, + true, + false, + false, + false, + nil, + ) + if err != nil { + return err + } + + return r.ch.QueueBind( + q.Name, + bindingKey, + exchange, + false, + nil, + ) +} + +func (r *RabbitMQ) PublishJSON(ctx context.Context, exchange string, routingKey string, payload any) error { + if r == nil || r.ch == nil { + return errors.New("rabbitmq is not initialized") + } + if exchange == "" || routingKey == "" { + return errors.New("exchange and routingKey are required") + } + b, err := json.Marshal(payload) + if err != nil { + return err + } + return r.ch.PublishWithContext(ctx, exchange, routingKey, false, false, amqp.Publishing{ + ContentType: "application/json", + DeliveryMode: amqp.Persistent, + Timestamp: time.Now(), + Body: b, + }) +} diff --git a/backend/internal/middleware/rabbitmq/socialMQ.go b/backend/internal/middleware/rabbitmq/socialMQ.go new file mode 100644 index 0000000..7d15140 --- /dev/null +++ b/backend/internal/middleware/rabbitmq/socialMQ.go @@ -0,0 +1,77 @@ +package rabbitmq + +import ( + "context" + "crypto/rand" + "encoding/hex" + "errors" + "time" +) + +type SocialMQ struct { + *RabbitMQ +} + +const ( + socialExchange = "social.events" + socialQueue = "social.events" + socialBindingKey = "social.*" + + socialFollowRK = "social.follow" + socialUnfollowRK = "social.unfollow" +) + +type SocialEvent struct { + EventID string `json:"event_id"` + Action string `json:"action"` + FollowerID uint `json:"follower_id"` + VloggerID uint `json:"vlogger_id"` + OccurredAt time.Time `json:"occurred_at"` +} + +func NewSocialMQ(base *RabbitMQ) (*SocialMQ, error) { + if base == nil { + return nil, errors.New("rabbitmq base is nil") + } + if err := base.DeclareTopic(socialExchange, socialQueue, socialBindingKey); err != nil { + return nil, err + } + return &SocialMQ{RabbitMQ: base}, nil +} + +func (s *SocialMQ) Follow(ctx context.Context, followerID, vloggerID uint) error { + return s.publish(ctx, "follow", socialFollowRK, followerID, vloggerID) +} + +func (s *SocialMQ) UnFollow(ctx context.Context, followerID, vloggerID uint) error { + return s.publish(ctx, "unfollow", socialUnfollowRK, followerID, vloggerID) +} + +func (s *SocialMQ) publish(ctx context.Context, action, routingKey string, followerID, vloggerID uint) error { + if s == nil || s.RabbitMQ == nil { + return errors.New("social mq is not initialized") + } + if followerID == 0 || vloggerID == 0 { + return errors.New("followerID and vloggerID are required") + } + id, err := newEventID(16) + if err != nil { + return err + } + evt := SocialEvent{ + EventID: id, + Action: action, + FollowerID: followerID, + VloggerID: vloggerID, + OccurredAt: time.Now().UTC(), + } + return s.PublishJSON(ctx, socialExchange, routingKey, evt) +} + +func newEventID(n int) (string, error) { + b := make([]byte, n) + if _, err := rand.Read(b); err != nil { + return "", err + } + return hex.EncodeToString(b), nil +}