40分钟学 Go 语言高并发：RPC服务开发实战

RPC服务开发实战

一、RPC服务基础概览

让我们通过一个完整的订单服务示例来学习RPC服务开发：

syntax = "proto3";package order;
option go_package = "./pb";// 订单服务定义
service OrderService {// 创建订单rpc CreateOrder (CreateOrderRequest) returns (CreateOrderResponse);// 查询订单rpc GetOrder (GetOrderRequest) returns (GetOrderResponse);// 更新订单状态rpc UpdateOrderStatus (UpdateOrderStatusRequest) returns (UpdateOrderStatusResponse);// 取消订单rpc CancelOrder (CancelOrderRequest) returns (CancelOrderResponse);// 订单状态流式更新rpc WatchOrderStatus (WatchOrderRequest) returns (stream OrderStatusUpdate);
}// 订单信息
message Order {string order_id = 1;string user_id = 2;repeated OrderItem items = 3;OrderStatus status = 4;double total_amount = 5;int64 created_at = 6;int64 updated_at = 7;
}// 订单项
message OrderItem {string product_id = 1;int32 quantity = 2;double price = 3;
}// 订单状态
enum OrderStatus {UNKNOWN = 0;PENDING = 1;PAID = 2;PROCESSING = 3;SHIPPED = 4;DELIVERED = 5;CANCELLED = 6;
}// 创建订单请求
message CreateOrderRequest {string user_id = 1;repeated OrderItem items = 2;
}// 创建订单响应
message CreateOrderResponse {Order order = 1;
}// 查询订单请求
message GetOrderRequest {string order_id = 1;
}// 查询订单响应
message GetOrderResponse {Order order = 1;
}// 更新订单状态请求
message UpdateOrderStatusRequest {string order_id = 1;OrderStatus status = 2;
}// 更新订单状态响应
message UpdateOrderStatusResponse {Order order = 1;
}// 取消订单请求
message CancelOrderRequest {string order_id = 1;string reason = 2;
}// 取消订单响应
message CancelOrderResponse {bool success = 1;string message = 2;
}// 监听订单状态请求
message WatchOrderRequest {string order_id = 1;
}// 订单状态更新
message OrderStatusUpdate {string order_id = 1;OrderStatus status = 2;string message = 3;int64 timestamp = 4;
}

现在实现服务端代码：

package mainimport ("context""fmt""log""net""sync""time"pb "your_project/pb""google.golang.org/grpc""google.golang.org/grpc/codes""google.golang.org/grpc/status"
)type orderServer struct {pb.UnimplementedOrderServiceServermu         sync.RWMutexorders     map[string]*pb.OrderstatusSubs map[string][]chan *pb.OrderStatusUpdate
}func newOrderServer() *orderServer {return &orderServer{orders:     make(map[string]*pb.Order),statusSubs: make(map[string][]chan *pb.OrderStatusUpdate),}
}// 生成订单ID
func generateOrderID() string {return fmt.Sprintf("ORD%d", time.Now().UnixNano())
}// 计算订单总金额
func calculateTotal(items []*pb.OrderItem) float64 {var total float64for _, item := range items {total += float64(item.Quantity) * item.Price}return total
}// 创建订单
func (s *orderServer) CreateOrder(ctx context.Context, req *pb.CreateOrderRequest) (*pb.CreateOrderResponse, error) {if req.UserId == "" {return nil, status.Error(codes.InvalidArgument, "user_id is required")}if len(req.Items) == 0 {return nil, status.Error(codes.InvalidArgument, "order items cannot be empty")}now := time.Now().Unix()order := &pb.Order{OrderId:     generateOrderID(),UserId:      req.UserId,Items:       req.Items,Status:      pb.OrderStatus_PENDING,TotalAmount: calculateTotal(req.Items),CreatedAt:   now,UpdatedAt:   now,}s.mu.Lock()s.orders[order.OrderId] = orders.mu.Unlock()return &pb.CreateOrderResponse{Order: order}, nil
}// 查询订单
func (s *orderServer) GetOrder(ctx context.Context, req *pb.GetOrderRequest) (*pb.GetOrderResponse, error) {s.mu.RLock()order, exists := s.orders[req.OrderId]s.mu.RUnlock()if !exists {return nil, status.Errorf(codes.NotFound, "order not found: %s", req.OrderId)}return &pb.GetOrderResponse{Order: order}, nil
}// 更新订单状态
func (s *orderServer) UpdateOrderStatus(ctx context.Context, req *pb.UpdateOrderStatusRequest) (*pb.UpdateOrderStatusResponse, error) {s.mu.Lock()defer s.mu.Unlock()order, exists := s.orders[req.OrderId]if !exists {return nil, status.Errorf(codes.NotFound, "order not found: %s", req.OrderId)}// 状态校验if !isValidStatusTransition(order.Status, req.Status) {return nil, status.Errorf(codes.FailedPrecondition, "invalid status transition from %v to %v", order.Status, req.Status)}order.Status = req.Statusorder.UpdatedAt = time.Now().Unix()// 通知订阅者s.notifyStatusUpdate(order)return &pb.UpdateOrderStatusResponse{Order: order}, nil
}// 取消订单
func (s *orderServer) CancelOrder(ctx context.Context, req *pb.CancelOrderRequest) (*pb.CancelOrderResponse, error) {s.mu.Lock()defer s.mu.Unlock()order, exists := s.orders[req.OrderId]if !exists {return nil, status.Errorf(codes.NotFound, "order not found: %s", req.OrderId)}if order.Status != pb.OrderStatus_PENDING && order.Status != pb.OrderStatus_PAID {return &pb.CancelOrderResponse{Success: false,Message: "order cannot be cancelled in current status",}, nil}order.Status = pb.OrderStatus_CANCELLEDorder.UpdatedAt = time.Now().Unix()// 通知订阅者s.notifyStatusUpdate(order)return &pb.CancelOrderResponse{Success: true,Message: "order cancelled successfully",}, nil
}// 监听订单状态
func (s *orderServer) WatchOrderStatus(req *pb.WatchOrderRequest, stream pb.OrderService_WatchOrderStatusServer) error {if req.OrderId == "" {return status.Error(codes.InvalidArgument, "order_id is required")}s.mu.RLock()_, exists := s.orders[req.OrderId]s.mu.RUnlock()if !exists {return status.Errorf(codes.NotFound, "order not found: %s", req.OrderId)}// 创建状态更新通道updates := make(chan *pb.OrderStatusUpdate, 10)s.mu.Lock()s.statusSubs[req.OrderId] = append(s.statusSubs[req.OrderId], updates)s.mu.Unlock()defer func() {s.mu.Lock()// 移除订阅subs := s.statusSubs[req.OrderId]for i, ch := range subs {if ch == updates {s.statusSubs[req.OrderId] = append(subs[:i], subs[i+1:]...)break}}s.mu.Unlock()close(updates)}()// 监听状态更新for update := range updates {if err := stream.Send(update); err != nil {return err}}return nil
}// 通知订阅者状态更新
func (s *orderServer) notifyStatusUpdate(order *pb.Order) {update := &pb.OrderStatusUpdate{OrderId:   order.OrderId,Status:    order.Status,Message:   fmt.Sprintf("Order status updated to %v", order.Status),Timestamp: time.Now().Unix(),}// 发送更新给所有订阅者for _, ch := range s.statusSubs[order.OrderId] {select {case ch <- update:default:// channel已满，跳过}}
}// 验证状态转换是否有效
func isValidStatusTransition(from, to pb.OrderStatus) bool {switch from {case pb.OrderStatus_PENDING:return to == pb.OrderStatus_PAID || to == pb.OrderStatus_CANCELLEDcase pb.OrderStatus_PAID:return to == pb.OrderStatus_PROCESSING || to == pb.OrderStatus_CANCELLEDcase pb.OrderStatus_PROCESSING:return to == pb.OrderStatus_SHIPPEDcase pb.OrderStatus_SHIPPED:return to == pb.OrderStatus_DELIVEREDdefault:return false}
}func main() {lis, err := net.Listen("tcp", ":50051")if err != nil {log.Fatalf("failed to listen: %v", err)}s := grpc.NewServer(grpc.UnaryInterceptor(orderInterceptor),)pb.RegisterOrderServiceServer(s, newOrderServer())log.Println("Order server starting on :50051...")if err := s.Serve(lis); err != nil {log.Fatalf("failed to serve: %v", err)}
}// 订单服务拦截器
func orderInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {start := time.Now()resp, err := handler(ctx, req)duration := time.Since(start)// 记录请求日志log.Printf("Method: %s, Duration: %v, Error: %v",info.FullMethod,duration,err,)return resp, err
}

让我们创建服务的工作流程图：

现在添加性能测试代码：

package mainimport ("context""log""sync""testing""time"pb "your_project/pb""google.golang.org/grpc""google.golang.org/grpc/credentials/insecure"
)func BenchmarkCreateOrder(b *testing.B) {// 建立连接conn, err := grpc.Dial(":50051", grpc.WithTransportCredentials(insecure.NewCredentials()))if err != nil {b.Fatalf("did not connect: %v", err)}defer conn.Close()client := pb.NewOrderServiceClient(conn)ctx := context.Background()// 准备测试数据testOrder := &pb.CreateOrderRequest{UserId: "test-user",Items: []*pb.OrderItem{{ProductId: "prod-1",Quantity: 1,Price:    99.99,},},}b.ResetTimer()b.RunParallel(func(pb *testing.PB) {for pb.Next() {_, err := client.CreateOrder(ctx, testOrder)if err != nil {b.Errorf("CreateOrder failed: %v", err)}}})
}// 并发压力测试
func TestOrderConcurrency(t *testing.T) {conn, err := grpc.Dial(":50051", grpc.WithTransportCredentials(insecure.NewCredentials()))if err != nil {t.Fatalf("did not connect: %v", err)}defer conn.Close()client := pb.NewOrderServiceClient(conn)ctx := context.Background()concurrentUsers := 100ordersPerUser := 10var wg sync.WaitGrouperrorChan := make(chan error, concurrentUsers*ordersPerUser)start := time.Now()for i := 0; i < concurrentUsers; i++ {wg.Add(1)go func(userID int) {defer wg.Done()for j := 0; j < ordersPerUser; j++ {order := &pb.CreateOrderRequest{UserId: fmt.Sprintf("user-%d", userID),Items: []*pb.OrderItem{{ProductId: fmt.Sprintf("prod-%d", j),Quantity: 1,Price:    99.99,},},}_, err := client.CreateOrder(ctx, order)if err != nil {errorChan <- err}}}(i)}// 等待所有goroutine完成wg.Wait()close(errorChan)duration := time.Since(start)totalOrders := concurrentUsers * ordersPerUserordersPerSecond := float64(totalOrders) / duration.Seconds()// 统计错误var errorCount intfor err := range errorChan {t.Errorf("Order creation error: %v", err)errorCount++}t.Logf("Performance Results:")t.Logf("Total Orders: %d", totalOrders)t.Logf("Total Time: %v", duration)t.Logf("Orders/Second: %.2f", ordersPerSecond)t.Logf("Error Rate: %.2f%%", float64(errorCount)/float64(totalOrders)*100)
}// 延迟测试
func TestOrderLatency(t *testing.T) {conn, err := grpc.Dial(":50051", grpc.WithTransportCredentials(insecure.NewCredentials()))if err != nil {t.Fatalf("did not connect: %v", err)}defer conn.Close()client := pb.NewOrderServiceClient(conn)ctx := context.Background()var latencies []time.DurationtestCount := 100for i := 0; i < testCount; i++ {order := &pb.CreateOrderRequest{UserId: "latency-test",Items: []*pb.OrderItem{{ProductId: "prod-1",Quantity: 1,Price:    99.99,},},}start := time.Now()_, err := client.CreateOrder(ctx, order)latency := time.Since(start)latencies = append(latencies, latency)if err != nil {t.Errorf("CreateOrder failed: %v", err)}time.Sleep(time.Millisecond * 100) // 避免过度压力}// 计算统计信息var totalLatency time.Durationvar maxLatency time.Durationvar minLatency = latencies[0]for _, l := range latencies {totalLatency += lif l > maxLatency {maxLatency = l}if l < minLatency {minLatency = l}}avgLatency := totalLatency / time.Duration(len(latencies))t.Logf("Latency Results:")t.Logf("Average Latency: %v", avgLatency)t.Logf("Max Latency: %v", maxLatency)t.Logf("Min Latency: %v", minLatency)
}// 负载测试监控器
type LoadTestMonitor struct {successCount int64errorCount   int64totalLatency time.Durationmu           sync.Mutex
}func (m *LoadTestMonitor) recordSuccess(latency time.Duration) {m.mu.Lock()defer m.mu.Unlock()m.successCount++m.totalLatency += latency
}func (m *LoadTestMonitor) recordError() {m.mu.Lock()defer m.mu.Unlock()m.errorCount++
}func (m *LoadTestMonitor) getStats() (int64, int64, time.Duration) {m.mu.Lock()defer m.mu.Unlock()avgLatency := time.Duration(0)if m.successCount > 0 {avgLatency = m.totalLatency / time.Duration(m.successCount)}return m.successCount, m.errorCount, avgLatency
}func TestLoadScenarios(t *testing.T) {conn, err := grpc.Dial(":50051", grpc.WithTransportCredentials(insecure.NewCredentials()))if err != nil {t.Fatalf("did not connect: %v", err)}defer conn.Close()client := pb.NewOrderServiceClient(conn)ctx := context.Background()scenarios := []struct {name           stringconcurrency   intduration     time.Duration}{{"Light Load", 10, 30 * time.Second},{"Medium Load", 50, 30 * time.Second},{"Heavy Load", 100, 30 * time.Second},}for _, scenario := range scenarios {t.Run(scenario.name, func(t *testing.T) {monitor := &LoadTestMonitor{}var wg sync.WaitGroup// 启动工作goroutinefor i := 0; i < scenario.concurrency; i++ {wg.Add(1)go func(workerID int) {defer wg.Done()start := time.Now()for time.Since(start) < scenario.duration {order := &pb.CreateOrderRequest{UserId: fmt.Sprintf("load-test-%d", workerID),Items: []*pb.OrderItem{{ProductId: "prod-1",Quantity: 1,Price:    99.99,},},}requestStart := time.Now()_, err := client.CreateOrder(ctx, order)if err != nil {monitor.recordError()} else {monitor.recordSuccess(time.Since(requestStart))}time.Sleep(time.Millisecond * 100) // 控制请求速率}}(i)}wg.Wait()success, errors, avgLatency := monitor.getStats()t.Logf("Scenario: %s", scenario.name)t.Logf("Success Count: %d", success)t.Logf("Error Count: %d", errors)t.Logf("Average Latency: %v", avgLatency)t.Logf("Requests/Second: %.2f", float64(success+errors)/scenario.duration.Seconds())})}
}

这里我们为订单服务实现了一套完整的性能测试方案，包括：

1. 基准测试

   • 测试单个API的性能
   • 并发测试
   • 可重复性测试

2. 并发压力测试

   • 模拟多用户场景
   • 测试系统稳定性
   • 性能数据收集

3. 延迟测试

   • 响应时间测试
   • 统计分析
   • 性能瓶颈定位

4. 负载测试

   • 不同负载场景
   • 系统极限测试
   • 性能监控

五、性能优化建议

1. 接口层面

   • 使用流式处理大数据
   • 批量处理请求
   • 合理设置超时

2. 服务层面

   • 使用连接池
   • 优化资源使用
   • 缓存热点数据

3. 系统层面

   • 调整系统参数
   • 监控资源使用
   • 及时处理故障

六、总结

本节课我们实现了一个完整的RPC订单服务，涵盖了：

1. 服务设计与实现

   • 接口定义
   • 业务逻辑
   • 错误处理

2. 性能测试与优化

   • 基准测试
   • 压力测试
   • 监控分析

3. 最佳实践

   • 代码组织
   • 错误处理
   • 性能优化

怎么样今天的内容还满意吗？再次感谢观众老爷的观看，关注GZH：凡人的AI工具箱，回复666，送您价值199的AI大礼包。最后，祝您早日实现财务自由，还请给个赞，谢谢！