Building Go Microservices with gRPC: Streaming, Interceptors, and Load Balancing

gRPC has become the standard inter-service communication protocol for Go microservices, offering strong typing via Protocol Buffers, bidirectional streaming, and excellent performance. This guide covers everything from defining your first protobuf service to production load balancing.

Why gRPC for Go Microservices

gRPC advantages over REST:

• Strong typing: Protocol Buffers enforce schemas at compile time
• Performance: Binary serialization vs. JSON text
• Streaming: Native support for server, client, and bidirectional streaming
• Code generation: Clients and servers generated from .proto files
• Interceptors: Middleware analogous to HTTP middleware

Defining Protobuf Services

// user.proto
syntax = "proto3";
package user.v1;
option go_package = "github.com/example/app/gen/user/v1;userv1";

import "google/protobuf/timestamp.proto";

message User {
  string id = 1;
  string name = 2;
  string email = 3;
  google.protobuf.Timestamp created_at = 4;
}

message GetUserRequest {
  string id = 1;
}

message CreateUserRequest {
  string name = 1;
  string email = 2;
}

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

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

// Streaming: watch for user updates
message WatchUsersRequest {
  repeated string ids = 1;
}

message UserEvent {
  enum Type {
    CREATED = 0;
    UPDATED = 1;
    DELETED = 2;
  }
  Type type = 1;
  User user = 2;
}

service UserService {
  rpc GetUser(GetUserRequest) returns (User);
  rpc CreateUser(CreateUserRequest) returns (User);
  rpc ListUsers(ListUsersRequest) returns (ListUsersResponse);
  
  // Server streaming: server sends multiple responses
  rpc WatchUsers(WatchUsersRequest) returns (stream UserEvent);
  
  // Client streaming: client sends multiple requests
  rpc BatchCreateUsers(stream CreateUserRequest) returns (ListUsersResponse);
  
  // Bidirectional streaming
  rpc Chat(stream ChatMessage) returns (stream ChatMessage);
}

Generate Go code:

protoc --go_out=. --go_opt=paths=source_relative \
       --go-grpc_out=. --go-grpc_opt=paths=source_relative \
       user.proto

Implementing the Server

package main

import (
    "context"
    "net"
    "log"
    
    "google.golang.org/grpc"
    "google.golang.org/grpc/codes"
    "google.golang.org/grpc/status"
    
    userv1 "github.com/example/app/gen/user/v1"
)

type UserServer struct {
    userv1.UnimplementedUserServiceServer
    store UserStore
}

func (s *UserServer) GetUser(ctx context.Context, req *userv1.GetUserRequest) (*userv1.User, error) {
    user, err := s.store.Get(ctx, req.Id)
    if err != nil {
        if errors.Is(err, ErrNotFound) {
            return nil, status.Errorf(codes.NotFound, "user %s not found", req.Id)
        }
        return nil, status.Errorf(codes.Internal, "failed to get user: %v", err)
    }
    return userToProto(user), nil
}

func (s *UserServer) WatchUsers(req *userv1.WatchUsersRequest, stream userv1.UserService_WatchUsersServer) error {
    ctx := stream.Context()
    eventCh := s.store.Subscribe(req.Ids)
    defer s.store.Unsubscribe(eventCh)
    
    for {
        select {
        case event, ok := <-eventCh:
            if !ok {
                return nil
            }
            if err := stream.Send(&userv1.UserEvent{
                Type: userv1.UserEvent_UPDATED,
                User: userToProto(event.User),
            }); err != nil {
                return err
            }
        case <-ctx.Done():
            return ctx.Err()
        }
    }
}

func (s *UserServer) BatchCreateUsers(stream userv1.UserService_BatchCreateUsersServer) error {
    var created []*userv1.User
    for {
        req, err := stream.Recv()
        if err == io.EOF {
            break
        }
        if err != nil {
            return status.Errorf(codes.Internal, "failed to receive: %v", err)
        }
        user, err := s.store.Create(stream.Context(), req.Name, req.Email)
        if err != nil {
            return status.Errorf(codes.Internal, "failed to create user: %v", err)
        }
        created = append(created, userToProto(user))
    }
    return stream.SendAndClose(&userv1.ListUsersResponse{Users: created})
}

func main() {
    lis, err := net.Listen("tcp", ":50051")
    if err != nil {
        log.Fatalf("failed to listen: %v", err)
    }
    
    server := grpc.NewServer(
        grpc.ChainUnaryInterceptor(
            loggingInterceptor,
            authInterceptor,
            recoveryInterceptor,
        ),
        grpc.ChainStreamInterceptor(
            streamLoggingInterceptor,
        ),
    )
    
    userv1.RegisterUserServiceServer(server, &UserServer{})
    
    log.Println("gRPC server listening on :50051")
    if err := server.Serve(lis); err != nil {
        log.Fatalf("failed to serve: %v", err)
    }
}

gRPC Interceptors

Interceptors provide middleware functionality for gRPC:

// Unary interceptor for logging
func loggingInterceptor(
    ctx context.Context,
    req interface{},
    info *grpc.UnaryServerInfo,
    handler grpc.UnaryHandler,
) (interface{}, error) {
    start := time.Now()
    resp, err := handler(ctx, req)
    log.Printf("%s completed in %v, err: %v", info.FullMethod, time.Since(start), err)
    return resp, err
}

// Authentication interceptor
func authInterceptor(
    ctx context.Context,
    req interface{},
    info *grpc.UnaryServerInfo,
    handler grpc.UnaryHandler,
) (interface{}, error) {
    md, ok := metadata.FromIncomingContext(ctx)
    if !ok {
        return nil, status.Error(codes.Unauthenticated, "missing metadata")
    }
    
    tokens := md.Get("authorization")
    if len(tokens) == 0 {
        return nil, status.Error(codes.Unauthenticated, "missing authorization token")
    }
    
    if !validateToken(tokens[0]) {
        return nil, status.Error(codes.Unauthenticated, "invalid token")
    }
    
    return handler(ctx, req)
}

// Recovery interceptor for panics
func recoveryInterceptor(
    ctx context.Context,
    req interface{},
    info *grpc.UnaryServerInfo,
    handler grpc.UnaryHandler,
) (resp interface{}, err error) {
    defer func() {
        if r := recover(); r != nil {
            log.Printf("panic in %s: %v", info.FullMethod, r)
            err = status.Errorf(codes.Internal, "internal server error")
        }
    }()
    return handler(ctx, req)
}

Error Handling with Status Codes

import (
    "google.golang.org/grpc/codes"
    "google.golang.org/grpc/status"
    "google.golang.org/genproto/googleapis/rpc/errdetails"
)

func (s *UserServer) CreateUser(ctx context.Context, req *userv1.CreateUserRequest) (*userv1.User, error) {
    // Validation with detailed error info
    if req.Email == "" {
        st, _ := status.New(codes.InvalidArgument, "email is required").
            WithDetails(&errdetails.BadRequest{
                FieldViolations: []*errdetails.BadRequest_FieldViolation{
                    {Field: "email", Description: "email cannot be empty"},
                },
            })
        return nil, st.Err()
    }
    
    user, err := s.store.Create(ctx, req.Name, req.Email)
    if err != nil {
        if errors.Is(err, ErrEmailConflict) {
            return nil, status.Errorf(codes.AlreadyExists, "email %s already exists", req.Email)
        }
        return nil, status.Errorf(codes.Internal, "create user: %v", err)
    }
    return userToProto(user), nil
}

Retry Logic with Exponential Backoff

import "google.golang.org/grpc/backoff"

func newClientWithRetry(addr string) (*grpc.ClientConn, error) {
    retryPolicy := `{
        "methodConfig": [{
            "name": [{"service": "user.v1.UserService"}],
            "retryPolicy": {
                "maxAttempts": 4,
                "initialBackoff": "0.1s",
                "maxBackoff": "1s",
                "backoffMultiplier": 2,
                "retryableStatusCodes": ["UNAVAILABLE", "RESOURCE_EXHAUSTED"]
            }
        }]
    }`
    
    return grpc.Dial(addr,
        grpc.WithTransportCredentials(insecure.NewCredentials()),
        grpc.WithDefaultServiceConfig(retryPolicy),
    )
}

Load Balancing

// Round-robin load balancing across multiple backends
func newLoadBalancedClient(endpoints []string) (*grpc.ClientConn, error) {
    // Use DNS-based service discovery
    return grpc.Dial(
        "dns:///user-service.default.svc.cluster.local:50051",
        grpc.WithTransportCredentials(insecure.NewCredentials()),
        grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy":"round_robin"}`),
    )
}

// Manual resolver for custom service discovery
type StaticResolver struct {
    endpoints []string
}

func (r *StaticResolver) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
    addrs := make([]resolver.Address, len(r.endpoints))
    for i, ep := range r.endpoints {
        addrs[i] = resolver.Address{Addr: ep}
    }
    cc.UpdateState(resolver.State{Addresses: addrs})
    return r, nil
}

Health Checking

import "google.golang.org/grpc/health/grpc_health_v1"

type HealthServer struct{}

func (h *HealthServer) Check(ctx context.Context, req *grpc_health_v1.HealthCheckRequest) (*grpc_health_v1.HealthCheckResponse, error) {
    // Check database, dependencies
    if err := checkDependencies(ctx); err != nil {
        return &grpc_health_v1.HealthCheckResponse{
            Status: grpc_health_v1.HealthCheckResponse_NOT_SERVING,
        }, nil
    }
    return &grpc_health_v1.HealthCheckResponse{
        Status: grpc_health_v1.HealthCheckResponse_SERVING,
    }, nil
}

func (h *HealthServer) Watch(req *grpc_health_v1.HealthCheckRequest, stream grpc_health_v1.Health_WatchServer) error {
    // Stream health updates
    ticker := time.NewTicker(5 * time.Second)
    defer ticker.Stop()
    for {
        select {
        case <-ticker.C:
            stream.Send(&grpc_health_v1.HealthCheckResponse{
                Status: grpc_health_v1.HealthCheckResponse_SERVING,
            })
        case <-stream.Context().Done():
            return nil
        }
    }
}

Conclusion

gRPC with Go provides a powerful foundation for microservice communication. Protocol Buffers give you type safety and efficient serialization. Interceptors handle cross-cutting concerns like auth and logging. Built-in streaming eliminates the need for WebSockets in most cases. Combined with proper error codes, retry logic, and health checking, you have everything needed to build production-grade microservice infrastructure.