Go微服务：微服务架构设计模式详解-编程实验室

Go微服务：微服务架构设计模式详解

1. 微服务架构概述

微服务架构是一种将单一应用程序划分为一组小服务的设计方法，每个服务运行在独立进程中，通过轻量级协议通信。微服务架构提供了更好的可扩展性、可维护性和技术灵活性。

2. 常见设计模式

2.1 微服务设计原则

单一职责：每个服务只负责一项业务功能
松耦合：服务之间通过接口通信，减少依赖
高内聚：相关功能放在同一个服务中
独立部署：每个服务可以独立部署和扩展

3. 聚合器模式

3.1 模式说明

聚合器模式在客户端或API网关层聚合多个服务的数据，返回组合结果。

3.2 Go实现

type OrderAggregator struct { orderClient pb.OrderServiceClient userClient pb.UserServiceClient productClient pb.ProductServiceClient } func (a *OrderAggregator) GetOrderDetail(ctx context.Context, orderID string) (*OrderDetail, error) { // 并发调用多个服务 var order *pb.Order var user *pb.User var products []*pb.Product var wg sync.WaitGroup var mu sync.Mutex var errs []error // 获取订单 wg.Add(1) go func() { defer wg.Done() var err error order, err = a.orderClient.GetOrder(ctx, &pb.GetOrderRequest{Id: orderID}) if err != nil { mu.Lock() errs = append(errs, err) mu.Unlock() } }() // 获取用户 wg.Add(1) go func() { defer wg.Done() // 获取订单后获取用户信息 if order != nil { var err error user, err = a.userClient.GetUser(ctx, &pb.GetUserRequest{Id: order.UserId}) if err != nil { mu.Lock() errs = append(errs, err) mu.Unlock() } } }() // 获取产品 wg.Add(1) go func() { defer wg.Done() if order != nil { var err error products, err = a.productClient.GetProductsByOrder(ctx, &pb.GetProductsRequest{OrderId: orderID}) if err != nil { mu.Lock() errs = append(errs, err) mu.Unlock() } } }() wg.Wait() if len(errs) > 0 { return nil, fmt.Errorf("failed to get order detail: %v", errs) } return &OrderDetail{ Order: order, User: user, Products: products, }, nil }

4. 代理模式

4.1 模式说明

代理模式通过API网关统一暴露服务，客户端只与网关通信，不直接访问后端服务。

4.2 Go实现

type APIProxy struct { router *gin.Engine services map[string]string // service name -> url } func NewAPIProxy() *APIProxy { proxy := &APIProxy{ router: gin.New(), services: make(map[string]string), } proxy.setupRoutes() return proxy } func (p *APIProxy) RegisterService(name, url string) { p.services[name] = url } func (p *APIProxy) setupRoutes() { p.router.GET("/api/users/:id", p.proxyTo("user-service")) p.router.POST("/api/users", p.proxyTo("user-service")) p.router.GET("/api/orders/:id", p.proxyTo("order-service")) p.router.POST("/api/orders", p.proxyTo("order-service")) } func (p *APIProxy) proxyTo(serviceName string) gin.HandlerFunc { return func(c *gin.Context) { url, ok := p.services[serviceName] if !ok { c.JSON(404, gin.H{"error": "service not found"}) return } // 修改请求路径和主机 path := c.Param("path") targetURL := url + path proxy := httputil.NewSingleHostReverseProxy(&url.URL{Scheme: "http", Host: url}) proxy.ServeHTTP(c.Writer, c.Request) } }

5. 链式模式

5.1 模式说明

链式模式将请求依次传递给多个服务，每个服务处理后传递给下一个。

func (s *ServiceChain) HandleRequest(ctx context.Context, req *Request) (*Response, error) { // 第一步：认证服务 authResp, err := s.authService.Authenticate(ctx, &AuthRequest{ Token: req.Token, }) if err != nil || !authResp.Valid { return nil, fmt.Errorf("authentication failed") } // 第二步：用户服务 userResp, err := s.userService.GetUser(ctx, &UserRequest{ UserID: authResp.UserID, }) if err != nil { return nil, fmt.Errorf("failed to get user: %w", err) } // 第三步：订单服务 orderResp, err := s.orderService.GetOrders(ctx, &OrderRequest{ UserID: userResp.User.ID, }) if err != nil { return nil, fmt.Errorf("failed to get orders: %w", err) } return &Response{ User: userResp.User, Orders: orderResp.Orders, }, nil }

6. 分支模式

6.1 模式说明

分支模式允许同时调用多个服务，根据结果选择响应。

func (s *BranchService) Search(ctx context.Context, query string) (*SearchResponse, error) { // 同时搜索多个服务 var wg sync.WaitGroup results := make(chan *SearchResult, 3) errors := make(chan error, 3) // 搜索用户 wg.Add(1) go func() { defer wg.Done() result, err := s.userSearch.Search(ctx, query) if err != nil { errors <- err return } results <- result }() // 搜索产品 wg.Add(1) go func() { defer wg.Done() result, err := s.productSearch.Search(ctx, query) if err != nil { errors <- err return } results <- result }() // 搜索订单 wg.Add(1) go func() { defer wg.Done() result, err := s.orderSearch.Search(ctx, query) if err != nil { errors <- err return } results <- result }() wg.Wait() close(results) close(errors) // 聚合结果 response := &SearchResponse{} for result := range results { response.Users = append(response.Users, result.Users...) response.Products = append(response.Products, result.Products...) response.Orders = append(response.Orders, result.Orders...) } return response, nil }

7. 微服务数据库设计

7.1 数据库per服务模式

每个微服务拥有独立的数据库：

// 用户服务 type UserRepository struct { db *sql.DB } func (r *UserRepository) FindByID(id string) (*User, error) { var user User err := r.db.QueryRow("SELECT id, name, email FROM users WHERE id = ?", id). Scan(&user.ID, &user.Name, &user.Email) if err != nil { return nil, err } return &user, nil } // 订单服务 type OrderRepository struct { db *sql.DB } func (r *OrderRepository) FindByUserID(userID string) ([]*Order, error) { rows, err := r.db.Query("SELECT id, user_id, total FROM orders WHERE user_id = ?", userID) if err != nil { return nil, err } defer rows.Close() var orders []*Order for rows.Next() { var order Order if err := rows.Scan(&order.ID, &order.UserID, &order.Total); err != nil { return nil, err } orders = append(orders, &order) } return orders, nil }

7.2 事件驱动数据共享

type UserEventHandler struct { kafkaProducer *kafka.Producer } func (h *UserEventHandler) HandleUserCreated(user *User) error { event := UserEvent{ Type: "USER_CREATED", UserID: user.ID, Payload: user, } data, err := json.Marshal(event) if err != nil { return err } return h.kafkaProducer.Produce(&kafka.Message{ Topic: "user-events", Key: []byte(user.ID), Value: data, }) } type OrderEventHandler struct { kafkaConsumer *kafka.Consumer } func (h *OrderEventHandler) Start() { go func() { for { msg, err := h.kafkaConsumer.Consume("user-events") if err != nil { continue } var event UserEvent if err := json.Unmarshal(msg.Value, &event); err != nil { continue } // 处理用户事件 if event.Type == "USER_CREATED" { h.syncUserData(event.Payload) } } }() }

8. Saga模式

8.1 模式说明

Saga模式通过一系列局部事务实现分布式事务，每个服务完成自己的部分，如果某一步失败，则执行补偿事务。

8.2 Saga实现

type CreateOrderSaga struct { steps []SagaStep } type SagaStep struct { Name string Execute func(ctx context.Context) error Compensate func(ctx context.Context) error } func NewCreateOrderSaga(userSvc UserService, orderSvc OrderService, paymentSvc PaymentService) *CreateOrderSaga { return &CreateOrderSaga{ steps: []SagaStep{ { Name: "validate_user", Execute: func(ctx context.Context) error { return userSvc.Validate(ctx, userID) }, Compensate: func(ctx context.Context) error { return nil // 无需补偿 }, }, { Name: "create_order", Execute: func(ctx context.Context) error { order, err := orderSvc.Create(ctx, orderRequest) if err != nil { return err } ctx.Set("order", order) return nil }, Compensate: func(ctx context.Context) error { order := ctx.Get("order").(*Order) return orderSvc.Cancel(ctx, order.ID) }, }, { Name: "process_payment", Execute: func(ctx context.Context) error { order := ctx.Get("order").(*Order) return paymentSvc.Charge(ctx, order.UserID, order.Total) }, Compensate: func(ctx context.Context) error { order := ctx.Get("order").(*Order) return paymentSvc.Refund(ctx, order.UserID, order.Total) }, }, }, } } func (s *CreateOrderSaga) Execute(ctx context.Context) error { for _, step := range s.steps { if err := step.Execute(ctx); err != nil { // 执行补偿 for i := len(s.steps) - 1; i >= 0; i-- { s.steps[i].Compensate(ctx) } return fmt.Errorf("saga failed at step %s: %w", step.Name, err) } } return nil }

9. 断路器模式

type CircuitBreaker struct { failureThreshold int successThreshold int timeout time.Duration state int failures int successes int lastFailure time.Time mu sync.RWMutex } func (cb *CircuitBreaker) Call(fn func() error) error { cb.mu.Lock() defer cb.mu.Unlock() if cb.state == StateOpen { if time.Since(cb.lastFailure) > cb.timeout { cb.state = StateHalfOpen cb.successes = 0 } else { return fmt.Errorf("circuit breaker is open") } } err := fn() if err != nil { cb.failures++ cb.lastFailure = time.Now() if cb.failures >= cb.failureThreshold { cb.state = StateOpen } return err } cb.successes++ if cb.successes >= cb.successThreshold { cb.state = StateClosed cb.failures = 0 } return nil }