Understanding gRPC: High‑Performance RPC with HTTP/2 and Protobuf

1. gRPC Overview

gRPC is a high‑performance, open‑source RPC framework developed by Google. It uses HTTP/2 for transport and Protocol Buffers (binary) for payload serialization.

1.2 gRPC vs REST

Protocol : REST uses HTTP/1.1, gRPC uses HTTP/2.

Data format : REST uses JSON/text, gRPC uses binary Protocol Buffers.

Code generation : REST typically relies on OpenAPI/Swagger, gRPC generates native code from .proto files.

Type safety : REST is weakly typed, gRPC provides strong typing.

Streaming support : REST requires polling or WebSocket, gRPC supports streaming natively.

Performance : REST is moderate, gRPC is high.

Browser support : REST works natively in browsers, gRPC requires grpc‑web.

1.3 Application Scenarios

Microservice communication – efficient inter‑service calls.

Mobile backend – low bandwidth, low latency.

Real‑time streaming – Streaming APIs.

Multi‑language microservices – cross‑language calls.

IoT communication – lightweight protocol.

2. Protocol Buffers

2.1 Definition

Protocol Buffers (Protobuf) is Google’s binary serialization format, smaller and faster than JSON.

// person.proto
syntax = "proto3";

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

2.2 Data‑type Mapping

double : Python float, Go float64, Java double float : Python float, Go float32, Java float int32 : Python int, Go int32, Java int int64 : Python int, Go int64, Java long bool : Python bool, Go bool, Java boolean string : Python str, Go string, Java String bytes : Python bytes, Go []byte, Java ByteString map : Python dict, Go map, Java Map repeated : Python list, Go slice, Java

List

2.3 Full Example

syntax = "proto3";
package user;
option go_package = "github.com/example/user;user";
option java_package = "com.example.user";
option java_multiple_files = true;

service UserService {
  rpc GetUser (GetUserRequest) returns (User);
  rpc CreateUser (CreateUserRequest) returns (User);
  rpc ListUsers (ListUsersRequest) returns (stream User);
  rpc BatchCreateUsers (stream CreateUserRequest) returns (stream CreateUserResponse);
}

message GetUserRequest { string user_id = 1; }
message CreateUserRequest { string name = 1; int32 age = 2; string email = 3; }
message ListUsersRequest { int32 page_size = 1; string page_token = 2; }
message User { string id = 1; string name = 2; int32 age = 3; string email = 4; int64 created_at = 5; }
message CreateUserResponse { User user = 1; bool success = 2; }

2.4 Code Generation

# Install protoc compiler
# macOS
brew install protobuf
# Ubuntu
sudo apt-get install protobuf-compiler

# Generate Python code
protoc --python_out=. person.proto

# Generate Go code
protoc --go_out=. --go-grpc_out=. person.proto

# Generate Java code
protoc --java_out=. person.proto

3. gRPC Communication Patterns

3.1 Unary Call

Client sends a single request and receives a single response.

class UserServiceServicer(user_service_pb2_grpc.UserServiceServicer):
    def GetUser(self, request, context):
        user = user_service_pb2.User(
            id=request.user_id,
            name="张三",
            age=30,
            email="[email protected]"
        )
        return user

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

def get_user():
    with grpc.insecure_channel('localhost:50051') as channel:
        stub = user_service_pb2_grpc.UserServiceStub(channel)
        response = stub.GetUser(user_service_pb2.GetUserRequest(user_id="12345"))
        print(f"User: {response.name}, Age: {response.age}")

3.2 Server Streaming

Client sends one request; server streams multiple responses.

def ListUsers(self, request, context):
    users = [
        user_service_pb2.User(id="1", name="张三", age=30),
        user_service_pb2.User(id="2", name="李四", age=25),
        user_service_pb2.User(id="3", name="王五", age=35),
    ]
    for user in users:
        yield user

def list_users():
    with grpc.insecure_channel('localhost:50051') as channel:
        stub = user_service_pb2_grpc.UserServiceStub(channel)
        for user in stub.ListUsers(user_service_pb2.ListUsersRequest(page_size=10)):
            print(f"User: {user.name}, Age: {user.age}")

3.3 Client Streaming

Client streams multiple requests; server returns a single response.

def BatchCreateUsers(self, request_iterator, context):
    created_count = 0
    for request in request_iterator:
        user = create_user_in_db(request)
        created_count += 1
    return user_service_pb2.BatchCreateResponse(success=True, created_count=created_count)

def batch_create_users():
    with grpc.insecure_channel('localhost:50051') as channel:
        stub = user_service_pb2_grpc.UserServiceStub(channel)
        requests = [
            user_service_pb2.CreateUserRequest(name="赵六", age=28),
            user_service_pb2.CreateUserRequest(name="钱七", age=32),
        ]
        response = stub.BatchCreateUsers(iter(requests))
        print(f"Created {response.created_count} users")

3.4 Bidirectional Streaming

Both client and server can send multiple messages independently.

async def ChatStream(self, request_iterator, context):
    async for request in request_iterator:
        response = user_service_pb2.ChatResponse(
            message=f"收到: {request.message}",
            timestamp=int(time.time())
        )
        yield response

async def chat_stream():
    async with grpc.aio.insecure_channel('localhost:50051') as channel:
        stub = user_service_pb2_grpc.ChatServiceStub(channel)
        response_stream = stub.ChatStream()
        for msg in ["你好", "今天天气如何", "再见"]:
            await response_stream.write(user_service_pb2.ChatRequest(message=msg))
        async for response in response_stream:
            print(f"收到回复: {response.message}")
        await response_stream.done_writing()

4. gRPC Service Definition

4.1 Full Proto Example (Order Service)

syntax = "proto3";
package order;
import "google/protobuf/timestamp.proto";
import "google/protobuf/empty.proto";
option go_package = "github.com/example/order;order";
option java_package = "com.example.order";
option java_multiple_files = true;

service OrderService {
  rpc CreateOrder (CreateOrderRequest) returns (CreateOrderResponse);
  rpc GetOrder (GetOrderRequest) returns (Order);
  rpc ListOrders (ListOrdersRequest) returns (stream Order);
  rpc CancelOrder (CancelOrderRequest) returns (google.protobuf.Empty);
}

enum OrderStatus {
  ORDER_STATUS_UNSPECIFIED = 0;
  ORDER_STATUS_PENDING = 1;
  ORDER_STATUS_PAID = 2;
  ORDER_STATUS_SHIPPED = 3;
  ORDER_STATUS_DELIVERED = 4;
  ORDER_STATUS_CANCELLED = 5;
}

message OrderItem {
  string product_id = 1;
  string product_name = 2;
  int32 quantity = 3;
  double price = 4;
}

message Order {
  string id = 1;
  string user_id = 2;
  repeated OrderItem items = 3;
  double total_amount = 4;
  OrderStatus status = 5;
  google.protobuf.Timestamp created_at = 6;
  google.protobuf.Timestamp updated_at = 7;
}

message CreateOrderRequest { string user_id = 1; repeated OrderItem items = 2; }
message CreateOrderResponse { Order order = 1; string message = 2; }
message GetOrderRequest { string order_id = 1; }
message ListOrdersRequest { string user_id = 1; int32 page_size = 2; string page_token = 3; }
message CancelOrderRequest { string order_id = 1; string reason = 2; }

4.2 gRPC Options Configuration

// Service‑level options (for gRPC‑gateway OpenAPI generation)
service OrderService {
  option (grpc.gateway.protoc_gen_openapiv2.options.openapiv2_service) = {
    summary: "订单服务 API";
    description: "提供订单创建、查询、取消等功能";
  };
}

// Method‑level HTTP mapping
rpc CreateOrder (CreateOrderRequest) returns (CreateOrderResponse) {
  option (google.api.http) = {
    post: "/v1/orders";
    body: "*";
  };
}

// Field‑level description and example
message Order {
  string id = 1 [(grpc.gateway.protoc_gen_openapiv2.options.openapiv2_field) = {
    description: "订单唯一标识符";
    example: "\"order_12345\"";
  }];
}

5. Interceptors and Middleware

5.1 Server Interceptors

import grpc, time, logging
from concurrent import futures

class LoggingInterceptor(grpc.UnaryUnaryServerInterceptor):
    """Logs method entry, duration and errors."""
    def intercept_service(self, continuation, handler_call_details):
        start = time.time()
        method = handler_call_details.method
        logging.info(f"Method {method} called")
        try:
            response = continuation(handler_call_details)
            elapsed = time.time() - start
            logging.info(f"Method {method} completed in {elapsed:.3f}s")
            return response
        except Exception as e:
            logging.error(f"Method {method} failed: {e}")
            raise

class AuthInterceptor(grpc.ServerInterceptor):
    """Simple token‑based authentication."""
    def __init__(self, valid_tokens):
        self.valid_tokens = valid_tokens
    def intercept_service(self, continuation, handler_call_details):
        metadata = dict(handler_call_details.invocation_metadata)
        token = metadata.get('authorization', '').replace('Bearer ', '')
        if token not in self.valid_tokens:
            context = handler_call_details.invocation_metadata
            grpc.ServicerContext.abort(grpc.StatusCode.UNAUTHENTICATED, 'Invalid token')
        return continuation(handler_call_details)

def serve_with_interceptors():
    server = grpc.server(
        futures.ThreadPoolExecutor(max_workers=10),
        interceptors=[LoggingInterceptor(), AuthInterceptor(valid_tokens={'token123'})]
    )
    order_service_pb2_grpc.add_OrderServiceServicer_to_server(OrderServiceImpl(), server)
    server.start()

5.2 Client Interceptors

import grpc, time, uuid, logging

class ClientLoggingInterceptor(grpc.UnaryUnaryClientInterceptor):
    """Collects request count, error count and logs latency."""
    def __init__(self):
        self.metrics = {'requests': 0, 'errors': 0}
    def intercept_unary_unary(self, continuation, client_call_details, request):
        self.metrics['requests'] += 1
        start = time.time()
        try:
            response = continuation(client_call_details, request)
            elapsed = time.time() - start
            logging.info(f"Request to {client_call_details.method} completed in {elapsed:.3f}s")
            return response
        except Exception as e:
            self.metrics['errors'] += 1
            logging.error(f"Request failed: {e}")
            raise

class ClientAuthInterceptor(grpc.UnaryUnaryClientInterceptor):
    """Adds Bearer token and request‑id metadata to each call."""
    def __init__(self, token):
        self.token = token
    def intercept_unary_unary(self, continuation, client_call_details, request):
        metadata = [
            ('authorization', f'Bearer {self.token}'),
            ('request-id', str(uuid.uuid4()))
        ]
        new_details = client_call_details._replace(invocation_metadata=metadata)
        return continuation(new_details, request)

def create_channel_with_interceptors():
    interceptors = [ClientLoggingInterceptor(), ClientAuthInterceptor(token='token123')]
    channel = grpc.intercept_channel(grpc.insecure_channel('localhost:50051'), *interceptors)
    return channel

5.3 Middleware Examples

class RateLimitInterceptor(grpc.ServerInterceptor):
    """Rejects requests when a token‑bucket limit is exceeded."""
    def __init__(self, max_requests_per_second):
        self.rate_limiter = TokenBucket(rate=max_requests_per_second)
    def _allow(self, continuation, handler_call_details):
        if not self.rate_limiter.try_acquire():
            context.abort(grpc.StatusCode.RESOURCE_EXHAUSTED, 'Rate limit exceeded')
        return continuation(handler_call_details)

class TracingInterceptor(grpc.ServerInterceptor):
    """Wraps each RPC in a tracing span."""
    def __init__(self, tracer):
        self.tracer = tracer
    def intercept_service(self, continuation, handler_call_details):
        with self.tracer.start_span(handler_call_details.method) as span:
            span.set_attribute('rpc.method', handler_call_details.method)
            try:
                response = continuation(handler_call_details)
                span.set_attribute('rpc.status', 'OK')
                return response
            except Exception as e:
                span.set_attribute('rpc.status', 'ERROR')
                span.record_exception(e)
                raise

6. Advanced Features

6.1 Load Balancing

# Client‑side weighted round‑robin balancer (custom policy)
class WeightedRoundRobinLoadBalancer(grpc.LoadBalancingPolicy):
    """Simple weighted round‑robin implementation."""
    def __init__(self, addresses):
        self.addresses = addresses
        self.current = 0
    def pick(self):
        address = self.addresses[self.current]
        self.current = (self.current + 1) % len(self.addresses)
        return address

# gRPC channel with built‑in round_robin and retries enabled
channel = grpc.insecure_channel(
    'localhost:50051',
    options=[
        ('grpc.lb_policy_name', 'round_robin'),
        ('grpc.enable_retries', 1),
    ]
)

6.2 Deadlines

# Server‑side deadline handling
def GetOrder(self, request, context):
    deadline = context.time_remaining()
    if deadline is None:
        deadline = 30  # default seconds
    if deadline < 1:
        context.abort(grpc.StatusCode.DEADLINE_EXCEEDED, 'Deadline too short')
    # ... business logic ...

# Client sets a 5‑second deadline
def get_order_with_deadline():
    with grpc.insecure_channel('localhost:50051') as channel:
        stub = order_service_pb2_grpc.OrderServiceStub(channel)
        try:
            response = stub.GetOrder(
                order_service_pb2.GetOrderRequest(order_id="123"),
                timeout=5.0
            )
        except grpc.RpcError as e:
            if e.code() == grpc.StatusCode.DEADLINE_EXCEEDED:
                print("Request timeout")

6.3 Error Handling

# Server returns structured errors
def GetOrder(self, request, context):
    order = find_order(request.order_id)
    if order is None:
        context.abort(grpc.StatusCode.NOT_FOUND, f'Order {request.order_id} not found')
    if not has_permission(request.order_id, context):
        context.abort(grpc.StatusCode.PERMISSION_DENIED, 'No permission to access this order')
    return order

# Client maps gRPC status codes to domain exceptions
def get_order(order_id):
    try:
        response = stub.GetOrder(order_service_pb2.GetOrderRequest(order_id=order_id))
        return response.order
    except grpc.RpcError as e:
        if e.code() == grpc.StatusCode.NOT_FOUND:
            raise OrderNotFoundError(order_id)
        elif e.code() == grpc.StatusCode.PERMISSION_DENIED:
            raise PermissionDeniedError()
        else:
            raise

6.4 Load‑Balancing Strategies

round_robin : distributes requests evenly.

pick_first : uses the first healthy endpoint (simple scenarios).

grpclb : server‑side load balancing for complex deployments.

weighted_round_robin : supports heterogeneous environments with weight‑based distribution.

7. Microservice Integration

7.1 gRPC‑Gateway

Expose gRPC services as REST endpoints by adding HTTP annotations to the .proto file and generating gateway code.

service OrderService {
  rpc CreateOrder (CreateOrderRequest) returns (Order) {
    option (google.api.http) = { post: "/v1/orders" body: "*" };
  }
  rpc GetOrder (GetOrderRequest) returns (Order) {
    option (google.api.http) = { get: "/v1/orders/{order_id}" };
  }
}

# Generate gateway and Swagger files
protoc \
    --grpc-gateway_out=. \
    --grpc-gateway_opt logtostderr=true \
    --swagger_out=. \
    person.proto

7.2 Service Discovery (Consul)

import consul, json
c = consul.Consul()

def register_service():
    c.agent.service.register(
        'order-service',
        service_id='order-service-1',
        address='localhost',
        port=50051,
        check=consul.Check.ttl('30s')
    )

def discover_service():
    _, services = c.health.service('order-service', passing=True)
    addresses = [f"{s['Service']['Address']}:{s['Service']['Port']}" for s in services]
    return addresses

addresses = discover_service()
channel = grpc.insecure_channel(
    f"{addresses[0]}",
    options=[('grpc.load_balancing_config', json.dumps({'round_robin': {}}))]
)

7.3 Health‑Check Service

class HealthCheckServicer(health_pb2_grpc.HealthServicer):
    """Standard gRPC health‑checking implementation."""
    def Check(self, request, context):
        return health_pb2.HealthCheckResponse(status=health_pb2.HealthCheckResponse.SERVING)
    def Watch(self, request, context):
        # Implementation omitted for brevity
        ...

def serve_with_health():
    server = grpc.server(futures.ThreadPoolExecutor())
    health_servicer = HealthCheckServicer()
    health_pb2_grpc.add_HealthServicer_to_server(health_servicer, server)
    order_service_pb2_grpc.add_OrderServiceServicer_to_server(OrderServiceImpl(), server)
    server.start()