feat(P3): Docker健康检查 + Worker优雅重启 + 前端错误监控

This commit is contained in:
Sisyphus
2026-04-25 16:07:58 +08:00
parent 025be6dd78
commit 3613cfe5a3
5 changed files with 468 additions and 404 deletions

View File

@@ -1,295 +1,315 @@
package main package main
import ( import (
"context" "context"
"feedsystem_video_go/internal/config" "feedsystem_video_go/internal/config"
"feedsystem_video_go/internal/db" "feedsystem_video_go/internal/db"
rediscache "feedsystem_video_go/internal/middleware/redis" rediscache "feedsystem_video_go/internal/middleware/redis"
"feedsystem_video_go/internal/observability" "feedsystem_video_go/internal/observability"
"feedsystem_video_go/internal/social" "feedsystem_video_go/internal/social"
"feedsystem_video_go/internal/video" "feedsystem_video_go/internal/video"
"feedsystem_video_go/internal/worker" "feedsystem_video_go/internal/worker"
"log" "log"
"os" "os"
"os/signal" "os/signal"
"strconv" "strconv"
"syscall" "syscall"
"time" "time"
amqp "github.com/rabbitmq/amqp091-go" amqp "github.com/rabbitmq/amqp091-go"
) "gorm.io/gorm"
)
const (
socialExchange = "social.events" const (
socialQueue = "social.events" socialExchange = "social.events"
socialBindingKey = "social.*" socialQueue = "social.events"
socialBindingKey = "social.*"
likeExchange = "like.events"
likeQueue = "like.events" likeExchange = "like.events"
likeBindingKey = "like.*" likeQueue = "like.events"
likeBindingKey = "like.*"
commentExchange = "comment.events"
commentQueue = "comment.events" commentExchange = "comment.events"
commentBindingKey = "comment.*" commentQueue = "comment.events"
commentBindingKey = "comment.*"
popularityExchange = "video.popularity.events"
popularityQueue = "video.popularity.events" popularityExchange = "video.popularity.events"
popularityBindingKey = "video.popularity.*" popularityQueue = "video.popularity.events"
) popularityBindingKey = "video.popularity.*"
)
func main() {
// 加载配置 func connectWithRetry(name string, maxRetries int, fn func() error) {
configPath := os.Getenv("CONFIG_PATH") for i := 0; i < maxRetries; i++ {
if configPath == "" { if err := fn(); err == nil {
configPath = "configs/config.yaml" return
} }
log.Printf("Loading config from %s", configPath) wait := time.Duration(1<<i) * time.Second
cfg, usedDefault, err := config.LoadLocalDev(configPath) if wait > 30*time.Second {
if err != nil { wait = 30 * time.Second
log.Fatalf("Failed to load config: %v", err) }
} log.Printf("%s 不可用,%v 后重试 (%d/%d)...", name, wait, i+1, maxRetries)
if usedDefault { time.Sleep(wait)
log.Printf("Config File %s not found, using default local config", configPath) }
} else { log.Fatalf("%s: 超过最大重试次数", name)
log.Printf("Config loaded from file: %s", configPath) }
}
// 连接数据库 func main() {
sqlDB, err := db.NewDB(cfg.Database) // 加载配置
if err != nil { configPath := os.Getenv("CONFIG_PATH")
log.Fatalf("Failed to connect database: %v", err) if configPath == "" {
} configPath = "configs/config.yaml"
defer db.CloseDB(sqlDB) }
log.Printf("Loading config from %s", configPath)
// 连接 Redis用于流行度更新 cfg, usedDefault, err := config.LoadLocalDev(configPath)
cache, err := rediscache.NewFromEnv(&cfg.Redis) if err != nil {
if err != nil { log.Fatalf("Failed to load config: %v", err)
log.Printf("Redis config error (popularity worker disabled): %v", err) }
cache = nil if usedDefault {
} else { log.Printf("Config File %s not found, using default local config", configPath)
pingCtx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond) } else {
defer cancel() log.Printf("Config loaded from file: %s", configPath)
if err := cache.Ping(pingCtx); err != nil { }
log.Printf("Redis not available (popularity worker disabled): %v", err) // 连接数据库(带重试)
_ = cache.Close() var sqlDB *gorm.DB
cache = nil connectWithRetry("MySQL", 10, func() error {
} else { var err error
defer cache.Close() sqlDB, err = db.NewDB(cfg.Database)
log.Printf("Redis connected (popularity worker enabled)") return err
} })
} defer db.CloseDB(sqlDB)
// 连接 RabbitMQ
url := "amqp://" + cfg.RabbitMQ.Username + ":" + cfg.RabbitMQ.Password + "@" + cfg.RabbitMQ.Host + ":" + strconv.Itoa(cfg.RabbitMQ.Port) + "/" // 连接 Redis用于流行度更新
conn, err := amqp.Dial(url) cache, err := rediscache.NewFromEnv(&cfg.Redis)
if err != nil { if err != nil {
log.Fatalf("Failed to connect rabbitmq: %v", err) log.Printf("Redis config error (popularity worker disabled): %v", err)
} cache = nil
defer conn.Close() } else {
// 创建 RabbitMQ 通道 pingCtx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond)
ch, err := conn.Channel() defer cancel()
if err != nil { if err := cache.Ping(pingCtx); err != nil {
log.Fatalf("Failed to open rabbitmq channel: %v", err) log.Printf("Redis not available (popularity worker disabled): %v", err)
} _ = cache.Close()
defer ch.Close() cache = nil
// 声明 Social 交换机和队列 } else {
if err := declareSocialTopology(ch); err != nil { defer cache.Close()
log.Fatalf("Failed to declare social topology: %v", err) log.Printf("Redis connected (popularity worker enabled)")
} }
if err := declareLikeTopology(ch); err != nil { }
log.Fatalf("Failed to declare like topology: %v", err) // 连接 RabbitMQ带重试
} url := "amqp://" + cfg.RabbitMQ.Username + ":" + cfg.RabbitMQ.Password + "@" + cfg.RabbitMQ.Host + ":" + strconv.Itoa(cfg.RabbitMQ.Port) + "/"
if err := declareCommentTopology(ch); err != nil { var conn *amqp.Connection
log.Fatalf("Failed to declare comment topology: %v", err) connectWithRetry("RabbitMQ", 10, func() error {
} var err error
if cache != nil { conn, err = amqp.Dial(url)
if err := declarePopularityTopology(ch); err != nil { return err
log.Fatalf("Failed to declare popularity topology: %v", err) })
} defer conn.Close()
} // 创建 RabbitMQ 通道
if err := ch.Qos(50, 0, false); err != nil { ch, err := conn.Channel()
log.Fatalf("Failed to set qos: %v", err) if err != nil {
} log.Fatalf("Failed to open rabbitmq channel: %v", err)
}
repo := social.NewSocialRepository(sqlDB) defer ch.Close()
socialWorker := worker.NewSocialWorker(ch, repo, socialQueue) // 声明 Social 交换机和队列
videoRepo := video.NewVideoRepository(sqlDB) if err := declareSocialTopology(ch); err != nil {
likeRepo := video.NewLikeRepository(sqlDB) log.Fatalf("Failed to declare social topology: %v", err)
commentRepo := video.NewCommentRepository(sqlDB) }
likeWorker := worker.NewLikeWorker(ch, likeRepo, videoRepo, likeQueue) if err := declareLikeTopology(ch); err != nil {
commentWorker := worker.NewCommentWorker(ch, commentRepo, videoRepo, commentQueue) log.Fatalf("Failed to declare like topology: %v", err)
var popularityWorker *worker.PopularityWorker }
if cache != nil { if err := declareCommentTopology(ch); err != nil {
popularityWorker = worker.NewPopularityWorker(ch, cache, popularityQueue) log.Fatalf("Failed to declare comment topology: %v", err)
} }
if cache != nil {
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) if err := declarePopularityTopology(ch); err != nil {
defer stop() log.Fatalf("Failed to declare popularity topology: %v", err)
}
pprofServer, err := observability.NewPprofServer( }
"Worker", if err := ch.Qos(50, 0, false); err != nil {
cfg.ObservabilityConfig.Pprof.Enabled, log.Fatalf("Failed to set qos: %v", err)
cfg.ObservabilityConfig.Pprof.WorkerAddr, }
)
if err != nil { repo := social.NewSocialRepository(sqlDB)
log.Printf("Failed to start worker pprof server: %v", err) socialWorker := worker.NewSocialWorker(ch, repo, socialQueue)
} videoRepo := video.NewVideoRepository(sqlDB)
if pprofServer != nil { likeRepo := video.NewLikeRepository(sqlDB)
defer pprofServer.Close() commentRepo := video.NewCommentRepository(sqlDB)
} likeWorker := worker.NewLikeWorker(ch, likeRepo, videoRepo, likeQueue)
commentWorker := worker.NewCommentWorker(ch, commentRepo, videoRepo, commentQueue)
errCh := make(chan error, 4) var popularityWorker *worker.PopularityWorker
log.Printf("Worker started, consuming queue=%s", socialQueue) if cache != nil {
go func() { errCh <- socialWorker.Run(ctx) }() popularityWorker = worker.NewPopularityWorker(ch, cache, popularityQueue)
log.Printf("Worker started, consuming queue=%s", likeQueue) }
go func() { errCh <- likeWorker.Run(ctx) }()
log.Printf("Worker started, consuming queue=%s", commentQueue) ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
go func() { errCh <- commentWorker.Run(ctx) }() defer stop()
if popularityWorker != nil {
log.Printf("Worker started, consuming queue=%s", popularityQueue) pprofServer, err := observability.NewPprofServer(
go func() { errCh <- popularityWorker.Run(ctx) }() "Worker",
} cfg.ObservabilityConfig.Pprof.Enabled,
cfg.ObservabilityConfig.Pprof.WorkerAddr,
err = <-errCh )
if err != nil && err != context.Canceled { if err != nil {
log.Fatalf("Worker stopped: %v", err) log.Printf("Failed to start worker pprof server: %v", err)
} }
log.Printf("Worker stopped") if pprofServer != nil {
} defer pprofServer.Close()
}
func declareSocialTopology(ch *amqp.Channel) error {
if err := ch.ExchangeDeclare( errCh := make(chan error, 4)
socialExchange, log.Printf("Worker started, consuming queue=%s", socialQueue)
"topic", go func() { errCh <- socialWorker.Run(ctx) }()
true, log.Printf("Worker started, consuming queue=%s", likeQueue)
false, go func() { errCh <- likeWorker.Run(ctx) }()
false, log.Printf("Worker started, consuming queue=%s", commentQueue)
false, go func() { errCh <- commentWorker.Run(ctx) }()
nil, if popularityWorker != nil {
); err != nil { log.Printf("Worker started, consuming queue=%s", popularityQueue)
return err go func() { errCh <- popularityWorker.Run(ctx) }()
} }
q, err := ch.QueueDeclare( err = <-errCh
socialQueue, if err != nil && err != context.Canceled {
true, log.Fatalf("Worker stopped: %v", err)
false, }
false, log.Printf("Worker stopped")
false, }
nil,
) func declareSocialTopology(ch *amqp.Channel) error {
if err != nil { if err := ch.ExchangeDeclare(
return err socialExchange,
} "topic",
true,
if err := ch.QueueBind( false,
q.Name, false,
socialBindingKey, false,
socialExchange, nil,
false, ); err != nil {
nil, return err
); err != nil { }
return err
} q, err := ch.QueueDeclare(
return nil socialQueue,
} true,
false,
func declarePopularityTopology(ch *amqp.Channel) error { false,
if err := ch.ExchangeDeclare( false,
popularityExchange, nil,
"topic", )
true, if err != nil {
false, return err
false, }
false,
nil, if err := ch.QueueBind(
); err != nil { q.Name,
return err socialBindingKey,
} socialExchange,
false,
q, err := ch.QueueDeclare( nil,
popularityQueue, ); err != nil {
true, return err
false, }
false, return nil
false, }
nil,
) func declarePopularityTopology(ch *amqp.Channel) error {
if err != nil { if err := ch.ExchangeDeclare(
return err popularityExchange,
} "topic",
true,
return ch.QueueBind( false,
q.Name, false,
popularityBindingKey, false,
popularityExchange, nil,
false, ); err != nil {
nil, return err
) }
}
q, err := ch.QueueDeclare(
func declareLikeTopology(ch *amqp.Channel) error { popularityQueue,
if err := ch.ExchangeDeclare( true,
likeExchange, false,
"topic", false,
true, false,
false, nil,
false, )
false, if err != nil {
nil, return err
); err != nil { }
return err
} return ch.QueueBind(
q.Name,
q, err := ch.QueueDeclare( popularityBindingKey,
likeQueue, popularityExchange,
true, false,
false, nil,
false, )
false, }
nil,
) func declareLikeTopology(ch *amqp.Channel) error {
if err != nil { if err := ch.ExchangeDeclare(
return err likeExchange,
} "topic",
true,
return ch.QueueBind( false,
q.Name, false,
likeBindingKey, false,
likeExchange, nil,
false, ); err != nil {
nil, return err
) }
}
q, err := ch.QueueDeclare(
func declareCommentTopology(ch *amqp.Channel) error { likeQueue,
if err := ch.ExchangeDeclare( true,
commentExchange, false,
"topic", false,
true, false,
false, nil,
false, )
false, if err != nil {
nil, return err
); err != nil { }
return err
} return ch.QueueBind(
q.Name,
q, err := ch.QueueDeclare( likeBindingKey,
commentQueue, likeExchange,
true, false,
false, nil,
false, )
false, }
nil,
) func declareCommentTopology(ch *amqp.Channel) error {
if err != nil { if err := ch.ExchangeDeclare(
return err commentExchange,
} "topic",
true,
return ch.QueueBind( false,
q.Name, false,
commentBindingKey, false,
commentExchange, nil,
false, ); err != nil {
nil, return err
) }
}
q, err := ch.QueueDeclare(
commentQueue,
true,
false,
false,
false,
nil,
)
if err != nil {
return err
}
return ch.QueueBind(
q.Name,
commentBindingKey,
commentExchange,
false,
nil,
)
}

View File

@@ -71,6 +71,11 @@ services:
condition: service_healthy condition: service_healthy
rabbitmq: rabbitmq:
condition: service_healthy condition: service_healthy
healthcheck:
test: ["CMD-SHELL", "wget -qO- --post-data='{}' --header='Content-Type: application/json' http://localhost:8080/account/findByID || exit 1"]
interval: 10s
timeout: 5s
retries: 3
worker: worker:
build: build:
@@ -87,6 +92,11 @@ services:
condition: service_healthy condition: service_healthy
rabbitmq: rabbitmq:
condition: service_healthy condition: service_healthy
healthcheck:
test: ["CMD-SHELL", "pgrep worker || exit 1"]
interval: 15s
timeout: 5s
retries: 3
frontend: frontend:
build: build:
@@ -97,6 +107,11 @@ services:
- "5173:80" - "5173:80"
depends_on: depends_on:
- backend - backend
healthcheck:
test: ["CMD-SHELL", "wget -qO- http://localhost:80/ || exit 1"]
interval: 10s
timeout: 5s
retries: 3
volumes: volumes:
mysql_data: mysql_data:

View File

@@ -1,99 +1,104 @@
import { useAuthStore } from '../stores/auth' import { useAuthStore } from '../stores/auth'
import { reportError } from '../utils/error-reporter'
export class ApiError extends Error {
status: number export class ApiError extends Error {
payload?: unknown status: number
payload?: unknown
constructor(message: string, status: number, payload?: unknown) {
super(message) constructor(message: string, status: number, payload?: unknown) {
this.name = 'ApiError' super(message)
this.status = status this.name = 'ApiError'
this.payload = payload this.status = status
} this.payload = payload
} }
}
type ApiErrorBody = { error?: string }
type ApiErrorBody = { error?: string }
const API_BASE = (import.meta.env.VITE_API_BASE as string | undefined) ?? '/api'
const API_BASE = (import.meta.env.VITE_API_BASE as string | undefined) ?? '/api'
export async function postJson<T>(path: string, body: unknown, options?: { authRequired?: boolean }): Promise<T> {
const auth = useAuthStore() export async function postJson<T>(path: string, body: unknown, options?: { authRequired?: boolean }): Promise<T> {
const token = auth.token const auth = useAuthStore()
const token = auth.token
if (options?.authRequired && !token) {
throw new ApiError('需要先登录(缺少 token', 401) if (options?.authRequired && !token) {
} throw new ApiError('需要先登录(缺少 token', 401)
}
const headers: Record<string, string> = { 'Content-Type': 'application/json' }
if (token) headers.Authorization = `Bearer ${token}` const headers: Record<string, string> = { 'Content-Type': 'application/json' }
if (token) headers.Authorization = `Bearer ${token}`
const res = await fetch(`${API_BASE}${path}`, {
method: 'POST', const res = await fetch(`${API_BASE}${path}`, {
headers, method: 'POST',
body: JSON.stringify(body ?? {}), headers,
}) body: JSON.stringify(body ?? {}),
})
const text = await res.text()
let data: unknown = null const text = await res.text()
if (text) { let data: unknown = null
try { if (text) {
data = JSON.parse(text) try {
} catch { data = JSON.parse(text)
data = text } catch {
} data = text
} }
}
if (!res.ok) {
if (res.status === 401) { if (!res.ok) {
auth.clearToken() if (res.status === 401) {
} auth.clearToken()
const msg = }
data && typeof data === 'object' && (data as ApiErrorBody).error const msg =
? String((data as ApiErrorBody).error) data && typeof data === 'object' && (data as ApiErrorBody).error
: `请求失败 (${res.status})` ? String((data as ApiErrorBody).error)
throw new ApiError(msg, res.status, data) : `请求失败 (${res.status})`
} const apiErr = new ApiError(msg, res.status, data)
reportError(apiErr, { path, status: res.status })
return data as T throw apiErr
} }
export async function postForm<T>(path: string, body: FormData, options?: { authRequired?: boolean }): Promise<T> { return data as T
const auth = useAuthStore() }
const token = auth.token
export async function postForm<T>(path: string, body: FormData, options?: { authRequired?: boolean }): Promise<T> {
if (options?.authRequired && !token) { const auth = useAuthStore()
throw new ApiError('需要先登录(缺少 token', 401) const token = auth.token
}
if (options?.authRequired && !token) {
const headers: Record<string, string> = {} throw new ApiError('需要先登录(缺少 token', 401)
if (token) headers.Authorization = `Bearer ${token}` }
const res = await fetch(`${API_BASE}${path}`, { const headers: Record<string, string> = {}
method: 'POST', if (token) headers.Authorization = `Bearer ${token}`
headers,
body, const res = await fetch(`${API_BASE}${path}`, {
}) method: 'POST',
headers,
const text = await res.text() body,
let data: unknown = null })
if (text) {
try { const text = await res.text()
data = JSON.parse(text) let data: unknown = null
} catch { if (text) {
data = text try {
} data = JSON.parse(text)
} } catch {
data = text
if (!res.ok) { }
if (res.status === 401) { }
auth.clearToken()
} if (!res.ok) {
const msg = if (res.status === 401) {
data && typeof data === 'object' && (data as ApiErrorBody).error auth.clearToken()
? String((data as ApiErrorBody).error) }
: `请求失败 (${res.status})` const msg =
throw new ApiError(msg, res.status, data) data && typeof data === 'object' && (data as ApiErrorBody).error
} ? String((data as ApiErrorBody).error)
: `请求失败 (${res.status})`
return data as T const apiErr = new ApiError(msg, res.status, data)
} reportError(apiErr, { path, status: res.status })
throw apiErr
}
return data as T
}

View File

@@ -1,10 +1,16 @@
import { createApp } from 'vue' import { createApp } from 'vue'
import { createPinia } from 'pinia' import { createPinia } from 'pinia'
import './style.css' import './style.css'
import App from './App.vue' import App from './App.vue'
import router from './router' import router from './router'
import { reportError } from './utils/error-reporter'
const app = createApp(App)
app.use(createPinia()) const app = createApp(App)
app.use(router) app.use(createPinia())
app.mount('#app') app.use(router)
app.config.errorHandler = (err, _instance, info) => {
reportError(err instanceof Error ? err : new Error(String(err)), { info })
}
app.mount('#app')

View File

@@ -0,0 +1,18 @@
export function reportError(error: Error, context?: Record<string, unknown>) {
if (import.meta.env.DEV) {
console.error('[ErrorReporter]', error.message, context)
return
}
fetch('/api/error-report', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
message: error.message,
stack: error.stack,
context,
timestamp: new Date().toISOString(),
}),
}).catch(() => {
/* 静默失败,避免错误上报自身导致循环 */
})
}