gRPC

# buf.gen.yaml
version: v2
inputs:
  - directory: api
plugins:
  - local: ["go", "run", "google.golang.org/protobuf/cmd/protoc-gen-go@v1.36.11"]
    out: api
    opt: paths=source_relative
  - local: ["go", "run", "google.golang.org/grpc/cmd/protoc-gen-go-grpc@v1.6.2"]
    out: api
    opt: paths=source_relative


# buf.yaml
version: v2
modules:
  - path: api
  - path: internal
deps:
  - buf.build/googleapis/googleapis

// api/hello/v1/hello.proto

syntax = "proto3";
     
package hello.v1;

option go_package = "grpcc/api/hello/v1;v1";

service TodoService {
    rpc Hello (HelloReq) returns (HelloResp) {}
}

message HelloReq {
    string name = 1;
}
message HelloResp {
    string message = 1;
}

// server/main.go

package main

import (
	v1 "grpcc/api/hello/v1"
	"log"
	"net"

	"google.golang.org/grpc"
	"google.golang.org/grpc/reflection"
)

const (
	// Address 监听地址
	Address string = ":9000"
	// Network 网络通信协议
	Network string = "tcp"
)

func main() {
	// 监听本地端口
	listener, err := net.Listen(Network, Address)
	if err != nil {
		log.Fatalf("net.Listen err: %v", err)
	}
	log.Println(Address + " net.Listing...")
	// 新建gRPC服务器实例
	grpcServer := grpc.NewServer()
	// 在gRPC服务器注册我们的服务
	v1.RegisterHelloServiceServer(grpcServer, &HelloServiceServer{})

	reflection.Register(grpcServer)
	//用服务器 Serve() 方法以及我们的端口信息区实现阻塞等待，直到进程被杀死或者 Stop() 被调用
	err = grpcServer.Serve(listener)
	if err != nil {
		log.Fatalf("grpcServer.Serve err: %v", err)
	}
}


// server/server.go
package main

import (
	"context"
	v1 "grpcc/api/hello/v1"
)

type HelloServiceServer struct {
	v1.UnimplementedHelloServiceServer
}

func (s *HelloServiceServer) Hello(ctx context.Context, req *v1.HelloReq) (*v1.HelloResp, error) {
	return &v1.HelloResp{
		Message: "Hello, " + req.GetName(),
	}, nil
}

// client/client.go

package main

import (
	"context"
	v1 "grpcc/api/hello/v1"
	"log"

	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
)

const (
	// Address 连接地址
	Address string = ":9000"
)

func main() {
	// 连接服务器
	conn, err := grpc.NewClient(Address, grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		log.Fatalf("net.Connect err: %v", err)
	}
	defer conn.Close()

	// 建立gRPC连接
	grpcClient := v1.NewHelloServiceClient(conn)
	// 创建发送结构体
	req := v1.HelloReq{
		Name: "grpc",
	}
	// 调用我们的服务(Route方法)
	// 同时传入了一个 context.Context ，在有需要时可以让我们改变RPC的行为，比如超时/取消一个正在运行的RPC
	res, err := grpcClient.Hello(context.Background(), &req)
	if err != nil {
		log.Fatalf("Call Route err: %v", err)
	}
	// 打印返回值
	log.Println(res)
}

// api/hello/v1/stock.proto

syntax = "proto3";
     
package hello.v1;

option go_package = "grpcc/api/hello/v1;v1";

service StockService {
    // （服务端流式 RPC）
    rpc GetStockInfo (StockReq) returns (stream StockResp) {}
}

message StockReq {
    string symbol = 1;
}
message StockResp {
    string name = 1;
    string price = 2;
    string change = 3;
    int64 timestamp = 4;
}

package main

import (
	"fmt"
	v1 "grpcc/api/hello/v1"
	"math/rand/v2"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

type StockServiceServer struct {
	v1.UnimplementedStockServiceServer
}

func (s *StockServiceServer) GetStockInfo(
	req *v1.StockReq,
	stream grpc.ServerStreamingServer[v1.StockResp],
) error {
	if req.GetSymbol() == "" {
		return status.Error(codes.InvalidArgument, "stock symbol is required")
	}

	// 模拟初始股票价格
	price := 100.0

	ticker := time.NewTicker(time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-stream.Context().Done():
			return stream.Context().Err()

		case <-ticker.C:
			// 模拟价格波动
			change := (rand.Float64() - 0.5) * 2
			price += change

			resp := &v1.StockResp{

				Price:     fmt.Sprint(price),
				Change:    fmt.Sprint(change),
				Timestamp: time.Now().Unix(),
			}

			if err := stream.Send(resp); err != nil {
				return err
			}
		}
	}
}

	gClient := v1.NewStockServiceClient(conn)
	stockReq := v1.StockReq{
		Symbol: "AAPL",
	}

	stockStream, err := gClient.GetStockInfo(context.Background(), &stockReq)
	if err != nil {
		log.Fatalf("Call GetStockInfo err: %v", err)
	}

	for {
		stockResp, err := stockStream.Recv()
		if err != nil {
			log.Fatalf("Call GetStockInfo err: %v", err)
		}
		log.Printf("Stock Name: %s, Price: %s, change: %s", stockResp.GetName(), stockResp.GetPrice(), stockResp.GetChange())
	}

// 增加 ctx
clientDeadline := time.Now().Add(time.Duration(3 * time.Second))
	ctx, cancel := context.WithDeadline(ctx, clientDeadline)
	defer cancel()

	res, err := grpcClient.Hello(context.Background(), &req)
	if err != nil {
		statu, ok := status.FromError(err)
		if ok {
			//判断是否为调用超时
			if statu.Code() == codes.DeadlineExceeded {
				log.Fatalln("Route timeout!")
			}
		}
		log.Fatalf("Call Route err: %v", err)
	}
	// 打印返回值
	log.Println(res)

func (s *StockServiceServer) GetStockInfo(ctx context.Context, req *v1.StockReq) (*v1.StockResp, error) {
    // 模拟耗时操作，期间周期性检查超时
    select {
    case <-ctx.Done():
        // 客户端 Deadline 到期或主动取消
        return nil, ctx.Err() // 框架会转换为 codes.DeadlineExceeded / codes.Canceled
    case <-time.After(2 * time.Second):
        // 正常业务处理完成
    }

    return &v1.StockResp{
        Symbol: req.GetSymbol(),
        Price:  100.0,
    }, nil
}

func UnaryServerTimeout(timeout time.Duration) grpc.UnaryServerInterceptor {
    return func(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
        if _, ok := ctx.Deadline(); !ok {
            var cancel context.CancelFunc
            ctx, cancel = context.WithTimeout(ctx, timeout)
            defer cancel()
        }
        return handler(ctx, req)
    }
}

// 注意：流式 RPC 不受 UnaryInterceptor 影响，需要单独用 grpc.StreamInterceptor 处理，逻辑类似
grpcServer := grpc.NewServer(
    grpc.UnaryInterceptor(UnaryServerTimeout(5 * time.Second)),
)

# 证书生成
openssl genrsa -out server.key 2048
openssl req -new -x509 -sha256 -key server.key -out server.pem -days 3650

func main() {
	// 监听本地端口
	listener, err := net.Listen(Network, Address)
	if err != nil {
		log.Fatalf("net.Listen err: %v", err)
	}
	log.Println(Address + " net.Listing...")
	// 新建gRPC服务器实例

	// 设置证书
	creds, err := credentials.NewServerTLSFromFile("cert/server.crt", "cert/server.key")
	if err != nil {
		log.Fatalf("credentials.NewServerTLSFromFile err: %v", err)
	}
	grpcServer := grpc.NewServer(grpc.Creds(creds))
	// 在gRPC服务器注册我们的服务
	v1.RegisterHelloServiceServer(grpcServer, &HelloServiceServer{})

	v1.RegisterStockServiceServer(grpcServer, &StockServiceServer{})

	reflection.Register(grpcServer)
	//用服务器 Serve() 方法以及我们的端口信息区实现阻塞等待，直到进程被杀死或者 Stop() 被调用
	err = grpcServer.Serve(listener)
	if err != nil {
		log.Fatalf("grpcServer.Serve err: %v", err)
	}
}

	creds, err := credentials.NewClientTLSFromFile("cert/server.crt", "grpcc")
	if err != nil {
		log.Fatalf("credentials.NewClientTLSFromFile err: %v", err)
	}
	conn, err := grpc.NewClient(Address, grpc.WithTransportCredentials(creds))
	if err != nil {
		log.Fatalf("net.Connect err: %v", err)
	}
	defer conn.Close()

### 创建根 CA
mkdir -p certs
cd certs

# 生成 CA 私钥
openssl genrsa -out ca.key 4096

# 生成 CA 根证书
openssl req -x509 -new -nodes \
  -key ca.key \
  -sha256 \
  -days 3650 \
  -out ca.crt \
  -subj "/CN=My gRPC Root CA"
  
  
### 生成服务端证书
openssl genrsa -out server.key 2048

openssl req -new \
  -key server.key \
  -out server.csr \
  -subj "/CN=localhost"
  
# 创建 server.ext
basicConstraints=CA:FALSE
keyUsage=digitalSignature,keyEncipherment
extendedKeyUsage=serverAuth
subjectAltName=DNS:localhost,IP:127.0.0.1

# 签发
openssl x509 -req \
  -in server.csr \
  -CA ca.crt \
  -CAkey ca.key \
  -CAcreateserial \
  -out server.crt \
  -days 825 \
  -sha256 \
  -extfile server.ext
  
  
# 生成客户端证书
openssl genrsa -out client.key 2048

openssl req -new \
  -key client.key \
  -out client.csr \
  -subj "/CN=grpc-client"
  
# 创建 client.ext
basicConstraints=CA:FALSE
keyUsage=digitalSignature,keyEncipherment
extendedKeyUsage=clientAuth

# 签发
openssl x509 -req \
  -in client.csr \
  -CA ca.crt \
  -CAkey ca.key \
  -CAcreateserial \
  -out client.crt \
  -days 825 \
  -sha256 \
  -extfile client.ext
  
  
# 检查证书
openssl verify -CAfile ca.crt server.crt
openssl verify -CAfile ca.crt client.crt

package security

import (
	"crypto/tls"
	"crypto/x509"
	"fmt"
	"os"

	"google.golang.org/grpc/credentials"
)

// ServerTLS 服务端双向 TLS 配置
func ServerTLS(
	certFile string,
	keyFile string,
	caFile string,
) (credentials.TransportCredentials, error) {
	// 加载服务端证书和私钥
	serverCert, err := tls.LoadX509KeyPair(certFile, keyFile)
	if err != nil {
		return nil, fmt.Errorf("load server certificate: %w", err)
	}

	// 加载可信 CA
	caPEM, err := os.ReadFile(caFile)
	if err != nil {
		return nil, fmt.Errorf("read CA certificate: %w", err)
	}

	clientCAs := x509.NewCertPool()
	if !clientCAs.AppendCertsFromPEM(caPEM) {
		return nil, fmt.Errorf("invalid client CA certificate")
	}

	tlsConfig := &tls.Config{
		Certificates: []tls.Certificate{serverCert},

		// 必须提供客户端证书，并且验证证书链
		ClientAuth: tls.RequireAndVerifyClientCert,
		ClientCAs:  clientCAs,

		MinVersion: tls.VersionTLS12,
	}

	return credentials.NewTLS(tlsConfig), nil
}

// ClientTLS 客户端双向 TLS 配置
func ClientTLS(
	certFile string,
	keyFile string,
	caFile string,
	serverName string,
) (credentials.TransportCredentials, error) {
	// 加载客户端证书和私钥
	clientCert, err := tls.LoadX509KeyPair(certFile, keyFile)
	if err != nil {
		return nil, fmt.Errorf("load client certificate: %w", err)
	}

	// 加载服务端 CA
	caPEM, err := os.ReadFile(caFile)
	if err != nil {
		return nil, fmt.Errorf("read CA certificate: %w", err)
	}

	roots := x509.NewCertPool()
	if !roots.AppendCertsFromPEM(caPEM) {
		return nil, fmt.Errorf("invalid server CA certificate")
	}

	tlsConfig := &tls.Config{
		Certificates: []tls.Certificate{clientCert},

		// 验证服务端证书
		RootCAs: roots,

		// 验证服务端证书 SAN 中的 DNS/IP
		ServerName: serverName,

		MinVersion: tls.VersionTLS12,
	}

	return credentials.NewTLS(tlsConfig), nil
}

import "github.com/grpc-ecosystem/go-grpc-middleware"
myServer := grpc.NewServer(
    // 流式RPC的拦截器
    grpc.StreamInterceptor(grpc_middleware.ChainStreamServer(
        grpc_ctxtags.StreamServerInterceptor(),
        grpc_opentracing.StreamServerInterceptor(),
        grpc_prometheus.StreamServerInterceptor,
        grpc_zap.StreamServerInterceptor(zapLogger),
        grpc_auth.StreamServerInterceptor(myAuthFunction),
        grpc_recovery.StreamServerInterceptor(),
    )),
    // 简单RPC的拦截器
    grpc.UnaryInterceptor(grpc_middleware.ChainUnaryServer(
        grpc_ctxtags.UnaryServerInterceptor(),
        grpc_opentracing.UnaryServerInterceptor(),
        grpc_prometheus.UnaryServerInterceptor,
        grpc_zap.UnaryServerInterceptor(zapLogger),
        grpc_auth.UnaryServerInterceptor(myAuthFunction),
        grpc_recovery.UnaryServerInterceptor(),
    )),
)

package main

import (
	"context"
	"fmt"

	"google.golang.org/grpc/metadata"
)

func main() {
	ctx := context.Background()
	// metadata key
	MDKey := "name"
	// 设置metadata数据
	metaData := map[string]string{
		MDKey: "jerry",
	}
	md := metadata.New(metaData)
	ctx = metadata.NewOutgoingContext(ctx, md)

	// 从ctx中获取metadata, get方法返回到是个slice,取第一个即可
	data := ""
	if md, ok := metadata.FromIncomingContext(ctx); ok {
		names := md.Get(MDKey)
		if len(names) > 0 {
			data = names[0]
		}
	}

	fmt.Printf("name: %v", data)
}