1
syntax = "proto3";
2

3
package user;
4

5
service UserService {
6
    //  unary RPC
7
    rpc GetUser(GetUserRequest) returns (User);
8

9
    // 服务端流式 RPC
10
    rpc ListUsers(ListRequest) returns (stream User);
11

12
    // 客户端流式 RPC
13
    rpc CreateUsers(stream CreateRequest) returns (CreateResponse);
14

15
    // 双向流式 RPC
16
    rpc Chat(stream Message) returns (stream Message);
17
}
18

19
message User {
20
    string id = 1;
21
    string name = 2;
22
    string email = 3;
23
}

1
// Proto 类型 -> Python/Go/TypeScript
2
int32      -> int/number
3
int64      -> int/Long
4
float      -> float/number
5
double     -> double/number
6
bool       -> bool/boolean
7
string     -> str/string
8
bytes      -> bytes/Uint8Array
9
repeated   -> list/array
10
map        -> dict/object

1
┌─────────────────────────────────────┐
2
│ Length (24-bit)                    │
3
├─────────────────────────────────────┤
4
│   Type (8-bit)  │  Flags (8-bit)   │
5
├─────────────────────────────────────┤
6
│         Stream ID (31-bit)         │
7
├─────────────────────────────────────┤
8
│            Payload                 │
9
└─────────────────────────────────────┘
10

11
# gRPC 帧类型
12
# 0x00: DATA
13
# 0x01: HEADERS
14
# 0x04: RESET_STREAM
15
# 0x07: GOAWAY

1
# Python 服务端
2
class UserServicer(user_pb2_grpc.UserServiceServicer):
3
    def GetUser(self, request, context):
4
        return user_pb2.User(
5
            id=request.id,
6
            name="张三",
7
            email="zhangsan@example.com"
8
        )

1
# 服务端流
2
def ListUsers(self, request, context):
3
    users = db.query_users(limit=request.limit)
4
    for user in users:
5
        yield user  # 流式返回

1
# 客户端流
2
def CreateUsers(self, request_iterator, context):
3
    created = []
4
    for request in request_iterator:
5
        user = db.create(request)
6
        created.append(user)
7
    return CreateResponse(users=created)

1
# 双向流
2
async def Chat(self, request_iterator, context):
3
    async for message in request_iterator:
4
        reply = await process_message(message)
5
        yield reply

1
# 客户端发送元数据
2
metadata = [("authorization", "Bearer token")]
3
response = stub.GetUser(request, metadata=metadata)
4

5
# 服务端读取元数据
6
def GetUser(self, request, context):
7
    token = context.invocation_metadata()
8
    auth = dict(token).get("authorization")
9
    if not auth:
10
        context.abort(grpc.StatusCode.UNAUTHENTICATED)

1
# 服务端拦截器
2
class AuthInterceptor(grpc.ServerInterceptor):
3
    def intercept_service(self, continuation, handler_call_details):
4
        meta = dict(handler_call_details.invocation_metadata)
5
        if not meta.get("authorization"):
6
            return grpc.unary_unary_rpc_method_handler(
7
                lambda *args: context.abort(grpc.StatusCode.UNAUTHENTICATED)
8
            )
9
        return continuation(handler_call_details)

1
# 服务端返回错误
2
context.abort(grpc.StatusCode.NOT_FOUND, "User not found")
3

4
# 客户端捕获
5
try:
6
    response = stub.GetUser(request)
7
except grpc.RpcError as e:
8
    print(f"错误: {e.code()} - {e.details()}")

1
import { grpc } from "@improbable-eng/grpc-web";
2

3
// 使用 grpc-web 调用
4
const client = grpc.invoke(UserService.GetUser, {
5
  request: request,
6
  host: "https://api.example.com",
7
  onMessage: user => {
8
    console.log(user);
9
  },
10
  onEnd: (code, message) => {},
11
});

1
# 服务端配置
2
server:
3
  credentials: grpc.ssl_server_credentials(
4
    server_cert,
5
    server_key
6
  )

1
# JWT 认证
2
class JWTInterceptor(grpc.ServerInterceptor):
3
    def intercept_service(self, continuation, details):
4
        meta = dict(details.invocation_metadata())
5
        token = meta.get("authorization", "").replace("Bearer ", "")
6
        if verify_jwt(token):
7
            return continuation(details)
8
        return lambda *args: abort(UNAUTHENTICATED)

1
# gRPC 负载均衡策略
2
channel = grpc.insecure_channel(
3
    "lb:///user-service",
4
    options=[
5
        ("grpc.load_balancing_policy", "round_robin")
6
    ]
7
)