gRPC 비밀: 기한, 시간 초과, 사용자 정의 컨텍스트 마스터하기

오픈 소스 원격 프로시저 호출(RPC) 프레임워크인 gRPC는 서비스 간 효율적이고 확장 가능한 통신을 지원합니다. gRPC의 중요한 측면 중 하나는 기한, 요청 시간 초과, 사용자 지정 구조를 포함한 컨텍스트 전파를 관리하는 것입니다.

이러한 메커니즘을 이해하면 서비스가 신속하게 응답하고, 합리적인 시간 프레임을 초과하는 작업에 리소스가 낭비되지 않으며, 사용자 지정 메타데이터가 효과적으로 전송되도록 할 수 있습니다.

기한 및 요청 시간 초과 이해

마감일

gRPC의 기한은 작업을 완료해야 하는 최대 시간을 지정합니다. 해당 기간 내에 작업이 완료되지 않으면 자동으로 종료됩니다. 응답이 없거나 느린 서비스로 인해 시스템 리소스가 무한정 묶여 있지 않도록 하려면 마감 기한이 필수적입니다.

요청 시간 초과

요청 시간 초과는 클라이언트가 서버의 응답을 기다리는 기간입니다. 이 기간 내에 서버가 응답하지 않으면 요청이 중단됩니다. 이 메커니즘은 클라이언트가 응답을 무기한 기다리지 않도록 보호합니다.

gRPC에서 기한 및 요청 시간 초과 설정

gRPC는 클라이언트 측과 서버 측 모두에서 기한을 설정하고 시간 초과를 요청하는 유연한 옵션을 제공합니다. Go에서 이를 수행하는 방법은 다음과 같습니다.

클라이언트 측 설정 기한

 import ( "context" "log" "time" "google.golang.org/grpc" pb "path/to/your/protobuf/package" ) func main() { conn, err := grpc.Dial("server_address", grpc.WithInsecure()) if err != nil { log.Fatalf("did not connect: %v", err) } defer conn.Close() client := pb.NewYourServiceClient(conn) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() resp, err := client.YourMethod(ctx, &pb.YourRequest{}) if err != nil { log.Fatalf("could not call method: %v", err) } log.Printf("Response: %v", resp) }

서버측 처리

서버 측에서 gRPC를 사용하면 기한을 적용하고 클라이언트가 지정한 기한이 초과되는 시나리오를 처리할 수 있습니다.

 import ( "context" "log" "net" "time" "google.golang.org/grpc" pb "path/to/your/protobuf/package" ) type server struct { pb.UnimplementedYourServiceServer } func (s *server) YourMethod(ctx context.Context, req *pb.YourRequest) (*pb.YourResponse, error) { select { case <-time.After(10 * time.Second): return &pb.YourResponse{}, nil case <-ctx.Done(): return nil, ctx.Err() } } func main() { lis, err := net.Listen("tcp", ":50051") if err != nil { log.Fatalf("failed to listen: %v", err) } s := grpc.NewServer() pb.RegisterYourServiceServer(s, &server{}) if err := s.Serve(lis); err != nil { log.Fatalf("failed to serve: %v", err) } }

상황에 맞는 사용자 정의 구조 전파

gRPC에서 컨텍스트를 통해 사용자 지정 구조를 보내려면 데이터를 컨텍스트에 연결하기 전에 직렬화한 다음 수신 측에서 역직렬화해야 합니다. 여기에는 사용자 정의 구조를 JSON 또는 프로토콜 버퍼와 같이 네트워크를 통해 전송할 수 있는 형식으로 변환한 다음 이 직렬화된 데이터를 컨텍스트 메타데이터에 추가하는 작업이 포함됩니다.

단계별 프로세스

1. 사용자 정의 구조 정의 : 보내려는 사용자 정의 구조를 정의합니다.
2. 구조 직렬화 : 사용자 정의 구조를 문자열 또는 바이트 배열로 변환합니다.
3. 컨텍스트에 연결 : 직렬화된 데이터를 컨텍스트 메타데이터에 추가합니다.
4. 전송 : 컨텍스트와 함께 gRPC 호출을 보냅니다.
5. 서버에서 추출 및 역직렬화 : 서버 측 컨텍스트에서 메타데이터를 추출하고 사용자 정의 구조로 다시 역직렬화합니다.

1단계: 사용자 정의 구조 정의

 type CustomStruct struct { Field1 string Field2 int }

2단계: 구조 직렬화

 import ( "context" "encoding/json" "fmt" "google.golang.org/grpc/metadata" ) func serializeCustomStruct(customStruct CustomStruct) (string, error) { data, err := json.Marshal(customStruct) if err != nil { return "", err } return string(data), nil }

3단계: 컨텍스트에 연결

 func attachCustomStructToContext(ctx context.Context, customStruct CustomStruct) (context.Context, error) { serializedData, err := serializeCustomStruct(customStruct) if err != nil { return nil, err } md := metadata.Pairs("custom-struct", serializedData) ctx = metadata.NewOutgoingContext(ctx, md) return ctx, nil }

4단계: 전송

 func main() { conn, err := grpc.Dial("localhost:50051", grpc.WithInsecure()) if err != nil { log.Fatalf("did not connect: %v", err) } defer conn.Close() client := pb.NewYourServiceClient(conn) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() customStruct := CustomStruct{Field1: "value1", Field2: 42} ctx, err = attachCustomStructToContext(ctx, customStruct) if err != nil { log.Fatalf("could not attach custom struct to context: %v", err) } resp, err := client.YourMethod(ctx, &pb.YourRequest{}) if err != nil { log.Fatalf("could not call method: %v", err) } log.Printf("Response: %v", resp) }

5단계: 서버에서 추출 및 역직렬화

 func deserializeCustomStruct(data string) (CustomStruct, error) { var customStruct CustomStruct err := json.Unmarshal([]byte(data), &customStruct) if err != nil { return CustomStruct{}, err } return customStruct, nil } func extractCustomStructFromContext(ctx context.Context) (CustomStruct, error) { md, ok := metadata.FromIncomingContext(ctx) if !ok { return CustomStruct{}, fmt.Errorf("no metadata found in context") } serializedData := md["custom-struct"] if len(serializedData) == 0 { return CustomStruct{}, fmt.Errorf("no custom struct found in metadata") } return deserializeCustomStruct(serializedData[0]) } func (s *server) YourMethod(ctx context.Context, req *pb.YourRequest) (*pb.YourResponse, error) { customStruct, err := extractCustomStructFromContext(ctx) if err != nil { return nil, err } log.Printf("Received custom struct: %+v", customStruct) select { case <-time.After(10 * time.Second): return &pb.YourResponse{}, nil case <-ctx.Done(): return nil, ctx.Err() } }

모든 gRPC 호출을 위한 미들웨어 구현

사용자 지정 구조를 포함한 컨텍스트 전파를 모든 gRPC 호출에서 일관되게 처리하려면 인터셉터를 사용할 수 있습니다. 인터셉터는 요청과 응답을 처리하고 로깅, 모니터링, 컨텍스트 메타데이터 처리와 같은 기능을 추가하는 미들웨어입니다.

단항 및 스트리밍 인터셉터

다양한 유형의 RPC 호출을 관리하려면 단항 인터셉터와 스트리밍 인터셉터가 모두 필요합니다.

• 단항 인터셉터 : 단일 요청-응답 주기를 처리합니다.

• 스트리밍 인터셉터 : 클라이언트 측 스트리밍, 서버 측 스트리밍, 양방향 스트리밍을 포함한 요청 및 응답 스트림을 처리합니다.

단항 인터셉터 구현

클라이언트측 단항 인터셉터:

 func unaryClientInterceptor( ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption, ) error { customStruct, ok := ctx.Value("customStruct").(CustomStruct) if ok { ctx, err := attachCustomStructToContext(ctx, customStruct) if err != nil { return err } } return invoker(ctx, method, req, reply, cc, opts...) }

서버측 단항 인터셉터:

 func unaryServerInterceptor( ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler, ) (interface{}, error) { customStruct, err := extractCustomStructFromContext(ctx) if err != nil { return nil, err } ctx = context.WithValue(ctx, "customStruct", customStruct) return handler(ctx, req) }

스트리밍 인터셉터 구현

클라이언트측 스트리밍 인터셉터:

 func streamClientInterceptor( ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption, ) (grpc.ClientStream, error) { customStruct, ok := ctx.Value("customStruct").(CustomStruct) if ok { ctx, err := attachCustomStructToContext(ctx, customStruct) if err != nil { return nil, err } } return

서버측 스트리밍 인터셉터:

 import ( "context" "google.golang.org/grpc" "google.golang.org/grpc/metadata" ) // StreamServerInterceptor handles server-side streaming func streamServerInterceptor( srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler, ) error { ctx := ss.Context() customStruct, err := extractCustomStructFromContext(ctx) if err != nil { return err } // Add custom struct to context for server handling newCtx := context.WithValue(ctx, "customStruct", customStruct) wrapped := grpc_middleware.WrapServerStream(ss) wrapped.WrappedContext = newCtx // Handle the request return handler(srv, wrapped) } // Example of using the interceptor in a gRPC server setup func main() { lis, err := net.Listen("tcp", ":50051") if err != nil { log.Fatalf("failed to listen: %v", err) } // Register the interceptors server := grpc.NewServer( grpc.UnaryInterceptor(unaryServerInterceptor), grpc.StreamInterceptor(streamServerInterceptor), ) // Register your gRPC service implementations here pb.RegisterYourServiceServer(server, &yourServiceServer{}) if err := server.Serve(lis); err != nil { log.Fatalf("failed to serve: %v", err) } }

단항 및 스트리밍 인터셉터를 생성하고 등록하면 사용자 지정 구조를 포함한 컨텍스트 전파가 모든 gRPC 호출에서 일관되게 처리되도록 할 수 있습니다. 이 접근 방식을 사용하면 사용자 지정 메타데이터가 적절하게 관리 및 전파되어 강력하고 유연한 gRPC 서비스를 구축할 수 있습니다.