-
Gin 路由与中间件（示例代码为简化版本，省略错误处理与依赖注入）
package main
import (
"context"
"net/http"
"os"
"time"
"github.com/gin-gonic/gin"
)
// 简化的服务接口
type OrderService interface {
CreateOrder(ctx context.Context, req CreateOrderRequest) (Order, error)
GetOrder(ctx context.Context, id string, opts QueryOptions) (AggregatedOrderView, error)
ListOrdersByUser(ctx context.Context, userId string, page, pageSize int) ([]Order, error)
}
// 请求与响应模型（示意）
type CreateOrderRequest struct {
UserId string `json:"userId" binding:"required"`
Items []OrderItem `json:"items" binding:"required"`
Amount float64 `json:"amount" binding:"required"`
Currency string `json:"currency" binding:"required"`
Metadata map[string]any `json:"metadata"`
IdempotencyKey string `json:"-"` // 从 Header 注入
}
type OrderItem struct {
SKU string `json:"sku"`
Quantity int `json:"quantity"`
}
type QueryOptions struct {
IncludeDetails bool
CacheOnly bool
}
func main() {
r := gin.New()
r.Use(gin.Recovery())
r.Use(correlationMiddleware()) // 注入/传播 X-Correlation-Id
r.Use(metricsMiddleware()) // 指标采集（示意）
r.Use(loggingMiddleware()) // 结构化日志（示意）
svc := newOrderService() // 注入真实实现
// 健康与就绪
r.GET("/health", func(c *gin.Context) { c.Status(http.StatusOK) })
r.GET("/ready", func(c *gin.Context) {
// 检查数据库、Kafka 客户端是否就绪（示意）
c.Status(http.StatusOK)
})
// 订单创建
r.POST("/orders", func(c *gin.Context) {
var req CreateOrderRequest
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request"})
return
}
req.IdempotencyKey = c.GetHeader("Idempotency-Key")
// 超时控制
ctx, cancel := context.WithTimeout(c.Request.Context(), 5*time.Second)
defer cancel()
order, err := svc.CreateOrder(ctx, req)
if err != nil {
// 根据错误类型返回 409/500 等（简化）
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusCreated, order)
})
// 单订单查询
r.GET("/orders/:id", func(c *gin.Context) {
id := c.Param("id")
include := c.DefaultQuery("includeDetails", "true") == "true"
cacheOnly := c.DefaultQuery("cacheOnly", "false") == "true"
opts := QueryOptions{IncludeDetails: include, CacheOnly: cacheOnly}
ctx, cancel := context.WithTimeout(c.Request.Context(), 2*time.Second)
defer cancel()
view, err := svc.GetOrder(ctx, id, opts)
if err != nil {
// 简化错误映射
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, view)
})
// 用户订单列表
r.GET("/orders", func(c *gin.Context) {
userId := c.Query("userId")
if userId == "" {
c.JSON(http.StatusBadRequest, gin.H{"error": "userId required"})
return
}
// 简化分页
page, pageSize := 1, 20
ctx, cancel := context.WithTimeout(c.Request.Context(), 2*time.Second)
defer cancel()
list, err := svc.ListOrdersByUser(ctx, userId, page, pageSize)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, gin.H{"data": list, "page": page, "pageSize": pageSize})
})
port := os.Getenv("SERVICE_PORT")
if port == "" {
port = "8080"
}
r.Run(":" + port)
}
// 以下中间件为示意，实际实现需结合日志与追踪框架
func correlationMiddleware() gin.HandlerFunc {
return func(c *gin.Context) {
corr := c.GetHeader("X-Correlation-Id")
if corr == "" {
corr = generateCorrelationID()
}
c.Set("correlationId", corr)
c.Writer.Header().Set("X-Correlation-Id", corr)
c.Next()
}
}
func metricsMiddleware() gin.HandlerFunc { return func(c *gin.Context) { c.Next() } }
func loggingMiddleware() gin.HandlerFunc { return func(c *gin.Context) { c.Next() } }
func generateCorrelationID() string { return time.Now().Format("20060102T150405.000000000") }
-
Saga 编排与事件处理（示意）
// 订单创建 Saga 简化伪代码
func (s *orderService) CreateOrder(ctx context.Context, req CreateOrderRequest) (Order, error) {
// 1. 幂等：检查/记录 Idempotency-Key
if exists, err := s.idemRepo.Exists(req.IdempotencyKey); err != nil {
return Order{}, err
} else if exists {
return s.idemRepo.GetOrderByKey(req.IdempotencyKey)
}
// 2. 创建订单（状态 PENDING）
order, err := s.orderRepo.CreatePending(req)
if err != nil { return Order{}, err }
// 3. 预留库存（带超时与熔断）
if err := s.cbInventory.Do(func() error {
return s.inventoryClient.Reserve(ctx, order.ID, req.Items)
}); err != nil {
// 补偿：取消订单
_ = s.orderRepo.MarkCancelled(order.ID, "inventory_reserve_failed")
return Order{}, err
}
_ = s.orderRepo.UpdateSagaState(order.ID, "INVENTORY_RESERVED")
// 4. 发起支付
if err := s.cbPayment.Do(func() error {
return s.paymentClient.Pay(ctx, order.ID, req.Amount, req.Currency)
}); err != nil {
// 补偿：释放库存 + 取消订单
_ = s.inventoryClient.Release(ctx, order.ID)
_ = s.orderRepo.MarkCancelled(order.ID, "payment_failed")
return Order{}, err
}
_ = s.orderRepo.UpdateSagaState(order.ID, "PAYMENT_INITIATED")
// 5. 发布 order.created 事件
_ = s.eventBus.Publish("order.created", Event{
Version: 1, IdempotencyKey: req.IdempotencyKey, CorrelationID: getCorr(ctx),
Data: map[string]any{"orderId": order.ID, "userId": req.UserId, "amount": req.Amount, "currency": req.Currency, "items": req.Items},
})
return s.orderRepo.Get(order.ID)
}
// 事件消费者：inventory.reserved / payment.succeeded
func (c *consumer) OnMessage(msg KafkaMessage) {
// 幂等检查：基于 idempotencyKey + eventVersion
if c.processed.IsProcessed(msg.IdempotencyKey, msg.EventVersion) { return }
switch msg.Type {
case "inventory.reserved":
// 更新订单状态为 INVENTORY_RESERVED（若支付已成功则推进完成）
_ = c.orderRepo.UpdateSagaState(msg.Data["orderId"].(string), "INVENTORY_RESERVED")
case "payment.succeeded":
orderID := msg.Data["orderId"].(string)
_ = c.orderRepo.UpdateStatus(orderID, "COMPLETED")
_ = c.notifyClient.Send(orderID) // 通知（示意）
}
c.processed.MarkProcessed(msg.IdempotencyKey, msg.EventVersion)
}
-
聚合查询与缓存/熔断（示意）
func (s *orderService) GetOrder(ctx context.Context, id string, opts QueryOptions) (AggregatedOrderView, error) {
base, err := s.orderRepo.Get(id)
if err != nil { return AggregatedOrderView{}, err }
if !opts.IncludeDetails {
return AggregatedOrderView{Order: base}, nil
}
if opts.CacheOnly {
if v, ok := s.cache.Get(id); ok { return v, nil }
return AggregatedOrderView{}, ErrCacheMiss
}
// 并行调用下游 + 超时 + 熔断
type result struct { name string; data any; err error }
ch := make(chan result, 3)
go func() { ch <- callWithCB(ctx, "inventory", s.cbInventory, func() (any, error) { return s.inventoryClient.View(ctx, id) }) }()
go func() { ch <- callWithCB(ctx, "payment", s.cbPayment, func() (any, error) { return s.paymentClient.View(ctx, id) }) }()
go func() { ch <- callWithCB(ctx, "notify", s.cbNotify, func() (any, error) { return s.notifyClient.View(ctx, id) }) }()
view := AggregatedOrderView{Order: base}
timeout := time.After(1500 * time.Millisecond)
completed := 0
for completed < 3 {
select {
case r := <-ch:
completed++
if r.err == nil {
switch r.name {
case "inventory": view.Inventory = r.data
case "payment": view.Payment = r.data
case "notify": view.Notification = r.data
}
}
case <-timeout:
// 超时：尝试返回缓存
if v, ok := s.cache.Get(id); ok { return v, nil }
return AggregatedOrderView{}, context.DeadlineExceeded
}
}
s.cache.Set(id, view, ttlFromConfig())
return view, nil
}
-
Kubernetes 示例（示意，需根据实际镜像与命名空间调整）
apiVersion: apps/v1
kind: Deployment
metadata:
name: order-aggregator
spec:
replicas: 2
selector:
matchLabels:
app: order-aggregator
template:
metadata:
labels:
app: order-aggregator
spec:
containers:
- name: app
image: your-registry/order-aggregator:latest
ports:
- containerPort: 8080
env:
- name: SERVICE_PORT
value: "8080"
- name: KAFKA_BROKERS
value: "kafka-0:9092,kafka-1:9092"
- name: DB_URL
valueFrom:
secretKeyRef:
name: order-db-secret
key: url
- name: CB_FAILURE_RATE
value: "0.5"
- name: RETRY_BACKOFF_MS
value: "200"
readinessProbe:
httpGet:
path: /ready
port: 8080
initialDelaySeconds: 5
livenessProbe:
httpGet:
path: /health
port: 8080
initialDelaySeconds: 5
---
apiVersion: v1
kind: Service
metadata:
name: order-aggregator
spec:
selector:
app: order-aggregator
ports:
- port: 80
targetPort: 8080
protocol: TCP
name: http
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: order-aggregator-hpa
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: order-aggregator
minReplicas: 2
maxReplicas: 10
metrics:
- type: Resource
resource:
name: cpu
target:
type: Utilization
averageUtilization: 70
