Protocol Buffers

Protocol Buffers(Protobuf, 프로토콜 버퍼)는 Google이 2008년 오픈소스로 공개한 언어 중립적, 플랫폼 중립적 데이터 직렬화 포맷입니다. 원래 Google 내부에서 효율적인 RPC 통신을 위해 개발되었으며, 현재는 gRPC의 기본 직렬화 방식으로 널리 사용됩니다. JSON이나 XML과 달리 이진 포맷으로 인코딩되어 데이터 크기가 작고 파싱 속도가 빠릅니다.

Protobuf의 핵심은 .proto 파일에 정의된 스키마입니다. 메시지 구조, 필드 타입, 필드 번호를 명시적으로 선언하고, protoc 컴파일러가 이를 다양한 언어(Python, Go, Java, C++, JavaScript 등)의 코드로 생성합니다. 필드 번호는 이진 인코딩에서 필드를 식별하는 데 사용되며, 한 번 지정하면 변경해서는 안 됩니다. 이를 통해 전후방 호환성(backward/forward compatibility)을 유지할 수 있습니다.

proto2와 proto3 두 가지 문법 버전이 있습니다. proto3는 2016년에 출시된 최신 버전으로, 기본값 개념을 단순화하고 required 필드를 제거했습니다. 모든 필드는 기본적으로 optional이며, 기본값이 설정되면 와이어에 전송되지 않아 메시지 크기가 더 작아집니다. 또한 proto3는 JSON 매핑을 공식 지원하여 REST API와의 상호 운용성이 향상되었습니다.

AI/ML 분야에서 Protobuf는 TensorFlow의 모델 저장 포맷(SavedModel), 학습 데이터 직렬화(TFRecord), 서빙 API(TensorFlow Serving)에 사용됩니다. 또한 Kubernetes, Envoy Proxy, Etcd 등 클라우드 네이티브 인프라 전반에서 핵심 통신 프로토콜로 채택되어 있습니다. JSON 대비 3-10배 작은 크기와 20-100배 빠른 파싱 성능으로, 대규모 분산 시스템에서 네트워크 비용과 지연 시간을 크게 줄일 수 있습니다.

.proto 스키마 정의

// user.proto
syntax = "proto3";

package example.user;

option go_package = "github.com/example/user/pb";
option java_package = "com.example.user";

import "google/protobuf/timestamp.proto";

// 사용자 메시지 정의
message User {
    int64 id = 1;                    // 필드 번호는 변경 불가
    string email = 2;
    string name = 3;
    UserStatus status = 4;
    repeated string roles = 5;       // 배열 타입
    Address address = 6;             // 중첩 메시지
    google.protobuf.Timestamp created_at = 7;

    // reserved로 삭제된 필드 번호 보호
    reserved 8, 9;
    reserved "old_field", "deprecated_field";
}

// 열거형 정의
enum UserStatus {
    USER_STATUS_UNSPECIFIED = 0;     // proto3는 0번 필수
    USER_STATUS_ACTIVE = 1;
    USER_STATUS_INACTIVE = 2;
    USER_STATUS_SUSPENDED = 3;
}

// 중첩 메시지
message Address {
    string street = 1;
    string city = 2;
    string country = 3;
    string postal_code = 4;
}

// gRPC 서비스 정의
service UserService {
    rpc GetUser(GetUserRequest) returns (User);
    rpc ListUsers(ListUsersRequest) returns (ListUsersResponse);
    rpc CreateUser(CreateUserRequest) returns (User);
    rpc UpdateUser(UpdateUserRequest) returns (User);
    rpc DeleteUser(DeleteUserRequest) returns (google.protobuf.Empty);

    // 스트리밍 RPC
    rpc WatchUsers(WatchUsersRequest) returns (stream User);
}

message GetUserRequest {
    int64 user_id = 1;
}

message ListUsersRequest {
    int32 page_size = 1;
    string page_token = 2;
    string filter = 3;
}

message ListUsersResponse {
    repeated User users = 1;
    string next_page_token = 2;
    int32 total_count = 3;
}

Python에서 Protobuf 사용

# protoc로 Python 코드 생성
# protoc --python_out=. --pyi_out=. user.proto

from google.protobuf.timestamp_pb2 import Timestamp
from google.protobuf.json_format import MessageToJson, Parse
from datetime import datetime
import user_pb2

# 메시지 생성
user = user_pb2.User()
user.id = 12345
user.email = "user@example.com"
user.name = "홍길동"
user.status = user_pb2.USER_STATUS_ACTIVE
user.roles.extend(["admin", "developer"])

# 중첩 메시지 설정
user.address.street = "테헤란로 123"
user.address.city = "서울"
user.address.country = "KR"

# Timestamp 설정
user.created_at.FromDatetime(datetime.now())

# 직렬화 (바이너리)
binary_data = user.SerializeToString()
print(f"Binary size: {len(binary_data)} bytes")  # ~100 bytes

# 역직렬화
loaded_user = user_pb2.User()
loaded_user.ParseFromString(binary_data)
print(f"Loaded user: {loaded_user.name}")

# JSON 변환 (디버깅/REST API 호환)
json_str = MessageToJson(user, preserving_proto_field_name=True)
print(json_str)
# {
#   "id": "12345",
#   "email": "user@example.com",
#   "name": "홍길동",
#   "status": "USER_STATUS_ACTIVE",
#   "roles": ["admin", "developer"],
#   "address": { ... }
# }

# JSON에서 파싱
user_from_json = Parse(json_str, user_pb2.User())

# 메시지 비교 및 병합
user2 = user_pb2.User()
user2.CopyFrom(user)
user2.MergeFrom(loaded_user)
print(user == user2)  # True

# oneof (택일 필드) 예제
# message Event {
#     oneof payload {
#         UserCreated user_created = 1;
#         UserUpdated user_updated = 2;
#         UserDeleted user_deleted = 3;
#     }
# }
# event.HasField("user_created")  # True/False
# event.WhichOneof("payload")     # "user_created"

gRPC 서비스 구현 (Python)

# gRPC 서버
import grpc
from concurrent import futures
import user_pb2
import user_pb2_grpc

class UserServiceServicer(user_pb2_grpc.UserServiceServicer):
    def __init__(self):
        self.users = {}  # 실제로는 DB 사용

    def GetUser(self, request, context):
        user_id = request.user_id
        if user_id not in self.users:
            context.abort(grpc.StatusCode.NOT_FOUND, f"User {user_id} not found")
        return self.users[user_id]

    def ListUsers(self, request, context):
        page_size = request.page_size or 10
        users = list(self.users.values())[:page_size]
        return user_pb2.ListUsersResponse(
            users=users,
            total_count=len(self.users)
        )

    def CreateUser(self, request, context):
        user = user_pb2.User()
        user.CopyFrom(request.user)
        user.id = len(self.users) + 1
        self.users[user.id] = user
        return user

    def WatchUsers(self, request, context):
        """서버 스트리밍 RPC"""
        for user in self.users.values():
            yield user

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    user_pb2_grpc.add_UserServiceServicer_to_server(
        UserServiceServicer(), server
    )
    server.add_insecure_port('[::]:50051')
    server.start()
    server.wait_for_termination()

# gRPC 클라이언트
def client_example():
    channel = grpc.insecure_channel('localhost:50051')
    stub = user_pb2_grpc.UserServiceStub(channel)

    # Unary RPC
    try:
        user = stub.GetUser(user_pb2.GetUserRequest(user_id=1))
        print(f"Got user: {user.name}")
    except grpc.RpcError as e:
        print(f"Error: {e.code()} - {e.details()}")

    # 스트리밍 RPC
    for user in stub.WatchUsers(user_pb2.WatchUsersRequest()):
        print(f"Received: {user.name}")

Go에서 Protobuf 사용

package main

import (
    "fmt"
    "log"

    "google.golang.org/protobuf/proto"
    "google.golang.org/protobuf/encoding/protojson"
    "google.golang.org/protobuf/types/known/timestamppb"

    pb "github.com/example/user/pb"
)

func main() {
    // 메시지 생성
    user := &pb.User{
        Id:     12345,
        Email:  "user@example.com",
        Name:   "홍길동",
        Status: pb.UserStatus_USER_STATUS_ACTIVE,
        Roles:  []string{"admin", "developer"},
        Address: &pb.Address{
            Street:  "테헤란로 123",
            City:    "서울",
            Country: "KR",
        },
        CreatedAt: timestamppb.Now(),
    }

    // 직렬화
    data, err := proto.Marshal(user)
    if err != nil {
        log.Fatal(err)
    }
    fmt.Printf("Binary size: %d bytes\n", len(data))

    // 역직렬화
    newUser := &pb.User{}
    if err := proto.Unmarshal(data, newUser); err != nil {
        log.Fatal(err)
    }

    // JSON 변환
    jsonBytes, _ := protojson.Marshal(user)
    fmt.Println(string(jsonBytes))

    // 깊은 복사
    clone := proto.Clone(user).(*pb.User)

    // 메시지 비교
    if proto.Equal(user, clone) {
        fmt.Println("Messages are equal")
    }
}