gRPC：构建高性能微服务

go install google.golang.org/protobuf/cmd/protoc-gen-go@latest
go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest

# 安装 Go gRPC 插件
go install google.golang.org/protobuf/cmd/protoc-gen-go@latest
go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest

# 安装 Protocol Buffers 编译器
# macOS
brew install protobuf

# Ubuntu
sudo apt install protobuf-compiler

# 验证安装
protoc --version

syntax = "proto3";

package user;

option go_package = "./user";

// 用户服务
service UserService {
  // 获取用户
  rpc GetUser (GetUserRequest) returns (User);

  // 创建用户
  rpc CreateUser (CreateUserRequest) returns (User);

  // 列出用户
  rpc ListUsers (ListUsersRequest) returns (stream User);

  // 更新用户信息（双向流）
  rpc UpdateUsers (stream UpdateUserRequest) returns (stream User);
}

// 用户消息
message User {
  int64 id = 1;
  string name = 2;
  string email = 3;
  int32 age = 4;
  int64 created_at = 5;
}

// 请求消息
message GetUserRequest {
  int64 id = 1;
}

message CreateUserRequest {
  string name = 1;
  string email = 2;
  int32 age = 3;
}

message ListUsersRequest {
  int32 page = 1;
  int32 page_size = 2;
}

message UpdateUserRequest {
  int64 id = 1;
  string name = 2;
  string email = 3;
  int32 age = 4;
}

protoc --go_out=. --go-grpc_out=. user.proto

package main

import (
	"context"
	"log"
	"net"
	"sync"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"

	pb "example/user"
)

type User struct {
	ID        int64
	Name      string
	Email     string
	Age       int32
	CreatedAt int64
}

type UserServiceServer struct {
	pb.UnimplementedUserServiceServer

	mu      sync.RWMutex
	users   map[int64]*User
	nextID  int64
}

func NewUserServiceServer() *UserServiceServer {
	return &UserServiceServer{
		users:  make(map[int64]*User),
		nextID: 1,
	}
}

// GetUser 实现获取用户
func (s *UserServiceServer) GetUser(ctx context.Context, req *pb.GetUserRequest) (*pb.User, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()

	user, ok := s.users[req.Id]
	if !ok {
		return nil, status.Errorf(codes.NotFound, "用户 %d 不存在", req.Id)
	}

	return &pb.User{
		Id:        user.ID,
		Name:      user.Name,
		Email:     user.Email,
		Age:       user.Age,
		CreatedAt: user.CreatedAt,
	}, nil
}

// CreateUser 实现创建用户
func (s *UserServiceServer) CreateUser(ctx context.Context, req *pb.CreateUserRequest) (*pb.User, error) {
	s.mu.Lock()
	defer s.mu.Unlock()

	user := &User{
		ID:        s.nextID,
		Name:      req.Name,
		Email:     req.Email,
		Age:       req.Age,
		CreatedAt: time.Now().Unix(),
	}

	s.users[user.ID] = user
	s.nextID++

	log.Printf("创建用户: %d (%s)", user.ID, user.Name)

	return &pb.User{
		Id:        user.ID,
		Name:      user.Name,
		Email:     user.Email,
		Age:       user.Age,
		CreatedAt: user.CreatedAt,
	}, nil
}

// ListUsers 实现列出用户（服务器流）
func (s *UserServiceServer) ListUsers(req *pb.ListUsersRequest, stream pb.UserService_ListUsersServer) error {
	s.mu.RLock()
	defer s.mu.RUnlock()

	count := 0
	for _, user := range s.users {
		if count >= int(req.PageSize) {
			break
		}

		err := stream.Send(&pb.User{
			Id:        user.ID,
			Name:      user.Name,
			Email:     user.Email,
			Age:       user.Age,
			CreatedAt: user.CreatedAt,
		})
		if err != nil {
			return err
		}

		count++
	}

	return nil
}

// UpdateUsers 实现批量更新（双向流）
func (s *UserServiceServer) UpdateUsers(stream pb.UserService_UpdateUsersServer) error {
	for {
		req, err := stream.Recv()
		if err != nil {
			return err
		}

		s.mu.Lock()
		user, ok := s.users[req.Id]
		if !ok {
			s.mu.Unlock()
			stream.Send(&pb.User{})
			continue
		}

		// 更新字段
		if req.Name != "" {
			user.Name = req.Name
		}
		if req.Email != "" {
			user.Email = req.Email
		}
		if req.Age > 0 {
			user.Age = req.Age
		}

		s.mu.Unlock()

		// 发送更新后的用户
		err = stream.Send(&pb.User{
			Id:        user.ID,
			Name:      user.Name,
			Email:     user.Email,
			Age:       user.Age,
			CreatedAt: user.CreatedAt,
		})
		if err != nil {
			return err
		}
	}
}

func main() {
	lis, err := net.Listen("tcp", ":50051")
	if err != nil {
		log.Fatalf("监听失败: %v", err)
	}

	s := grpc.NewServer()
	pb.RegisterUserServiceServer(s, NewUserServiceServer())

	log.Println("gRPC 服务器启动在 :50051")
	if err := s.Serve(lis); err != nil {
		log.Fatalf("服务失败: %v", err)
	}
}

package main

import (
	"context"
	"io"
	"log"
	"time"

	"google.golang.org/grpc"

	pb "example/user"
)

func main() {
	// 连接服务器
	conn, err := grpc.Dial("localhost:50051", grpc.WithInsecure())
	if err != nil {
		log.Fatalf("连接失败: %v", err)
	}
	defer conn.Close()

	client := pb.NewUserServiceClient(conn)
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	// 1. 创建用户
	log.Println("=== 创建用户 ===")
	user1, err := client.CreateUser(ctx, &pb.CreateUserRequest{
		Name:  "张三",
		Email: "zhangsan@example.com",
		Age:   25,
	})
	if err != nil {
		log.Fatalf("创建用户失败: %v", err)
	}
	log.Printf("创建成功: ID=%d, Name=%s", user1.Id, user1.Name)

	user2, err := client.CreateUser(ctx, &pb.CreateUserRequest{
		Name:  "李四",
		Email: "lisi@example.com",
		Age:   30,
	})
	if err != nil {
		log.Fatalf("创建用户失败: %v", err)
	}
	log.Printf("创建成功: ID=%d, Name=%s", user2.Id, user2.Name)

	// 2. 获取用户
	log.Println("\n=== 获取用户 ===")
	user, err := client.GetUser(ctx, &pb.GetUserRequest{Id: user1.Id})
	if err != nil {
		log.Fatalf("获取用户失败: %v", err)
	}
	log.Printf("用户信息: %+v", user)

	// 3. 列出用户（服务器流）
	log.Println("\n=== 列出用户 ===")
	stream, err := client.ListUsers(ctx, &pb.ListUsersRequest{
		Page:     1,
		PageSize: 10,
	})
	if err != nil {
		log.Fatalf("列出用户失败: %v", err)
	}

	for {
		user, err := stream.Recv()
		if err == io.EOF {
			break
		}
		if err != nil {
			log.Fatalf("接收失败: %v", err)
		}
		log.Printf("用户: ID=%d, Name=%s", user.Id, user.Name)
	}

	// 4. 批量更新（双向流）
	log.Println("\n=== 批量更新 ===")
	updateStream, err := client.UpdateUsers(ctx)
	if err != nil {
		log.Fatalf("更新用户失败: %v", err)
	}

	// 发送更新请求
	updates := []*pb.UpdateUserRequest{
		{Id: user1.Id, Name: "张三（已更新）", Age: 26},
		{Id: user2.Id, Email: "lisi_new@example.com"},
	}

	for _, update := range updates {
		err := updateStream.Send(update)
		if err != nil {
			log.Fatalf("发送更新失败: %v", err)
		}

		// 接收更新后的用户
		user, err := updateStream.Recv()
		if err != nil {
			log.Fatalf("接收更新失败: %v", err)
		}
		log.Printf("更新成功: ID=%d, Name=%s, Email=%s", user.Id, user.Name, user.Email)
	}

	updateStream.CloseSend()
}

// 服务器拦截器
func loggingInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
	start := time.Now()
	log.Printf("开始处理: %s", info.FullMethod)

	resp, err := handler(ctx, req)

	duration := time.Since(start)
	log.Printf("完成处理: %s (%v)", info.FullMethod, duration)

	return resp, err
}

// 客户端拦截器
func clientLoggingInterceptor(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
	start := time.Now()
	log.Printf("客户端调用: %s", method)

	err := invoker(ctx, method, req, reply, cc, opts...)

	duration := time.Since(start)
	log.Printf("客户端完成: %s (%v)", method, duration)

	return err
}

// 使用拦截器
func main() {
	// 服务器
	s := grpc.NewServer(
		grpc.UnaryInterceptor(loggingInterceptor),
	)

	// 客户端
	conn, err := grpc.Dial(
		"localhost:50051",
		grpc.WithInsecure(),
		grpc.WithUnaryInterceptor(clientLoggingInterceptor),
	)
}

// 从 context 获取传入的 metadata
md, ok := metadata.FromIncomingContext(ctx)
if ok {
	requestID := md.Get("x-request-id")
	log.Println("请求 ID:", requestID)
}

// 客户端发送 metadata
ctx := metadata.AppendToOutgoingContext(ctx, "x-request-id", "12345")
resp, err := client.SomeMethod(ctx, req)

// 服务器返回 header
header := metadata.New(map[string]string{
	"x-custom-header": "value",
})
grpc.SendHeader(ctx, header)

// 启用健康检查服务
import "google.golang.org/grpc/health"
import "google.golang.org/grpc/health/grpc_health_v1"

func main() {
	s := grpc.NewServer()

	// 注册健康检查服务
	healthServer := health.NewServer()
	grpc_health_v1.RegisterHealthServer(s, healthServer)

	// 设置服务状态
	healthServer.SetServingStatus("user.UserService", grpc_health_v1.HealthCheckResponse_SERVING)
}

// order_service.go - 订单服务调用用户服务
package main

import (
	"context"
	"log"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"

	userpb "example/user"
)

type OrderService struct {
	userClient userpb.UserServiceClient
}

func NewOrderService(userAddr string) (*OrderService, error) {
	conn, err := grpc.Dial(userAddr,
		grpc.WithInsecure(),
		grpc.WithBlock(),
		grpc.WithTimeout(5*time.Second),
	)
	if err != nil {
		return nil, err
	}
	return &OrderService{
		userClient: userpb.NewUserServiceClient(conn),
	}, nil
}

func (s *OrderService) CreateOrder(ctx context.Context, userID int64, amount float64) error {
	// 调用用户服务验证用户
	userCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
	defer cancel()

	_, err := s.userClient.GetUser(userCtx, &userpb.GetUserRequest{Id: userID})
	if err != nil {
		st, ok := status.FromError(err)
		if ok && st.Code() == codes.NotFound {
			return status.Errorf(codes.InvalidArgument, "用户 %d 不存在", userID)
		}
		return status.Errorf(codes.Internal, "验证用户失败: %v", err)
	}

	// 创建订单逻辑...
	log.Printf("为用户 %d 创建订单, 金额: %.2f", userID, amount)
	return nil
}

package main

import (
	"bufio"
	"context"
	"io"
	"log"
	"os"
	"time"

	"google.golang.org/grpc"
)

// proto 定义
// service FileService {
//   rpc Upload(stream FileChunk) returns (UploadStatus);
// }
// message FileChunk {
//   bytes data = 1;
//   string filename = 2;
//   int64 sequence = 3;
// }
// message UploadStatus {
//   bool success = 1;
//   string message = 2;
//   int64 total_bytes = 3;
// }

func uploadFile(client pb.FileServiceClient, filepath string) error {
	file, err := os.Open(filepath)
	if err != nil {
		return err
	}
	defer file.Close()

	stream, err := client.Upload(context.Background())
	if err != nil {
		return err
	}

	reader := bufio.NewReader(file)
	buffer := make([]byte, 1024*64) // 64KB 分片
	sequence := int64(0)

	for {
		n, err := reader.Read(buffer)
		if n > 0 {
			sequence++
			chunk := &pb.FileChunk{
				Data:     buffer[:n],
				Filename: filepath,
				Sequence: sequence,
			}
			if err := stream.Send(chunk); err != nil {
				return err
			}
		}
		if err == io.EOF {
			break
		}
		if err != nil {
			return err
		}
	}

	status, err := stream.CloseAndRecv()
	if err != nil {
		return err
	}

	log.Printf("上传结果: %s, 总字节: %d", status.Message, status.TotalBytes)
	return nil
}

// 服务器拦截器：验证 Token
func authInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
	// 从 metadata 获取 token
	md, ok := metadata.FromIncomingContext(ctx)
	if !ok {
		return nil, status.Error(codes.Unauthenticated, "缺少 metadata")
	}

	tokens := md.Get("authorization")
	if len(tokens) == 0 {
		return nil, status.Error(codes.Unauthenticated, "缺少 token")
	}

	// 验证 token
	token := tokens[0]
	userID, err := validateToken(token)
	if err != nil {
		return nil, status.Error(codes.Unauthenticated, "无效的 token")
	}

	// 将用户信息放入 context
	ctx = context.WithValue(ctx, "userID", userID)

	return handler(ctx, req)
}

// 客户端：发送 Token
type tokenAuth struct {
	token string
}

func (t *tokenAuth) GetRequestMetadata(ctx context.Context, uri ...string) (map[string]string, error) {
	return map[string]string{
		"authorization": t.token,
	}, nil
}

func (t *tokenAuth) RequireTransportSecurity() bool {
	return false
}

// 使用
conn, err := grpc.Dial(
	"localhost:50051",
	grpc.WithInsecure(),
	grpc.WithPerRPCCredentials(&tokenAuth{token: "your-token"}),
)

package main

import (
	"sync"
	"time"
)

// CircuitBreaker 简单熔断器实现
type CircuitBreaker struct {
	failureCount  int
	failureLimit  int
	timeout       time.Duration
	lastFailureTime time.Time
	state         string // closed, open, half-open
	mu            sync.Mutex
}

func NewCircuitBreaker(limit int, timeout time.Duration) *CircuitBreaker {
	return &CircuitBreaker{
		failureLimit: limit,
		timeout:      timeout,
		state:        "closed",
	}
}

func (cb *CircuitBreaker) Call(fn func() error) error {
	cb.mu.Lock()
	defer cb.mu.Unlock()

	if cb.state == "open" {
		if time.Since(cb.lastFailureTime) > cb.timeout {
			cb.state = "half-open"
			cb.failureCount = 0
		} else {
			return ErrCircuitOpen
		}
	}

	err := fn()
	if err != nil {
		cb.failureCount++
		cb.lastFailureTime = time.Now()
		if cb.failureCount >= cb.failureLimit {
			cb.state = "open"
		}
		return err
	}

	// 成功则重置
	cb.failureCount = 0
	cb.state = "closed"
	return nil
}

ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
resp, err := client.GetUser(ctx, req)

grpc.WithKeepaliveParams(keepalive.ClientParameters{
    Time:                10 * time.Second,
    Timeout:             3 * time.Second,
    PermitWithoutStream: true,
})

s := grpc.NewServer(
    grpc.MaxRecvMsgSize(16*1024*1024),  // 16MB
    grpc.MaxSendMsgSize(16*1024*1024),  // 16MB
)

func main() {
	lis, err := net.Listen("tcp", ":50051")
	if err != nil {
		log.Fatalf("监听失败: %v", err)
	}

	s := grpc.NewServer()
	pb.RegisterUserServiceServer(s, NewUserServiceServer())

	// 优雅关闭
	go func() {
		c := make(chan os.Signal, 1)
		signal.Notify(c, syscall.SIGINT, syscall.SIGTERM)
		<-c
		log.Println("正在优雅关闭...")
		s.GracefulStop()
	}()

	log.Println("gRPC 服务器启动在 :50051")
	if err := s.Serve(lis); err != nil {
		log.Fatalf("服务失败: %v", err)
	}
}

package main

import (
    "context"
    "fmt"
    "time"
    
    "google.golang.org/grpc"
)

// UnaryInterceptor 一元拦截器
type UnaryInterceptor func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error)

// LoggingInterceptor 日志拦截器
func LoggingInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
    start := time.Now()
    resp, err := handler(ctx, req)
    fmt.Printf("[%s] %s %v err=%v\n", time.Since(start), info.FullMethod, req, err)
    return resp, err
}

// AuthInterceptor 认证拦截器
func AuthInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
    // 从 metadata 获取 token
    md, ok := metadata.FromIncomingContext(ctx)
    if !ok {
        return nil, fmt.Errorf("missing metadata")
    }
    
    token := md.Get("authorization")
    if len(token) == 0 {
        return nil, fmt.Errorf("missing authorization token")
    }
    
    // 验证 token（简化）
    if token[0] != "Bearer valid-token" {
        return nil, fmt.Errorf("invalid token")
    }
    
    return handler(ctx, req)
}

func main() {
    server := grpc.NewServer(
        grpc.UnaryInterceptor(LoggingInterceptor),
    )
    _ = server
}

package main

import (
    "context"
    "fmt"
    
    "google.golang.org/grpc"
    healthpb "google.golang.org/grpc/health/grpc_health_v1"
)

type HealthServer struct {
    healthpb.UnimplementedHealthServer
}

func (s *HealthServer) Check(ctx context.Context, req *healthpb.HealthCheckRequest) (*healthpb.HealthCheckResponse, error) {
    return &healthpb.HealthCheckResponse{
        Status: healthpb.HealthCheckResponse_SERVING,
    }, nil
}

func (s *HealthServer) Watch(req *healthpb.HealthCheckRequest, stream healthpb.Health_WatchServer) error {
    return stream.Send(&healthpb.HealthCheckResponse{
        Status: healthpb.HealthCheckResponse_SERVING,
    })
}

func main() {
    fmt.Println("健康检查服务实现示例")
}

peer, ok := peer.FromContext(ctx)
if ok {
    fmt.Println("Client IP:", peer.Addr.String())
}

package main

import (
    "context"
    "fmt"
    "time"
)

// ServerStreamExample 服务端流式示例
type ServerStreamExample struct{}

func (s *ServerStreamExample) StreamData(req *DataRequest, stream DataStream_Server) error {
    for i := 0; i < 10; i++ {
        if err := stream.Send(&DataResponse{Value: int64(i)}); err != nil {
            return err
        }
        time.Sleep(100 * time.Millisecond)
    }
    return nil
}

type DataRequest struct{}
type DataResponse struct{ Value int64 }
type DataStream_Server interface{ Send(*DataResponse) error }

func main() {
    fmt.Println("服务端流式：服务端连续发送多个响应消息")
}

package main

import (
    "fmt"
    "io"
)

type ChatMessage struct {
    User    string
    Content string
}

type ChatStream interface {
    Send(*ChatMessage) error
    Recv() (*ChatMessage, error)
}

func ChatHandler(stream ChatStream) error {
    for {
        msg, err := stream.Recv()
        if err == io.EOF {
            return nil
        }
        if err != nil {
            return err
        }
        
        response := &ChatMessage{
            User:    "Server",
            Content: fmt.Sprintf("收到 %s 的消息: %s", msg.User, msg.Content),
        }
        if err := stream.Send(response); err != nil {
            return err
        }
    }
}

func main() {
    fmt.Println("双向流式：聊天服务器示例")
}

// 一元拦截器：处理普通请求
func unaryInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
    start := time.Now()
    md, _ := metadata.FromIncomingContext(ctx)
    log.Printf("RPC: %s, metadata: %v", info.FullMethod, md)
    resp, err := handler(ctx, req)
    log.Printf("RPC %s finished in %v", info.FullMethod, time.Since(start))
    return resp, err
}

// 流拦截器：处理流式请求
func streamInterceptor(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
    log.Printf("Stream RPC: %s", info.FullMethod)
    return handler(srv, ss)
}

// 链式使用
server := grpc.NewServer(
    grpc.ChainUnaryInterceptor(loggingInterceptor, authInterceptor, recoveryInterceptor),
    grpc.ChainStreamInterceptor(streamLoggingInterceptor),
)

// 服务端读取 metadata
func (s *Server) GetUser(ctx context.Context, req *pb.GetUserRequest) (*pb.User, error) {
    md, ok := metadata.FromIncomingContext(ctx)
    if ok {
        traceID := md.Get("x-trace-id")
        log.Printf("trace_id: %v", traceID)
    }
    // ...
}

// 客户端发送 metadata
ctx := metadata.NewOutgoingContext(context.Background(),
    metadata.Pairs("x-trace-id", "abc-123", "x-request-id", "req-456"))
resp, err := client.GetUser(ctx, req)