grpc 和限流Sentinel

基于gRPC的微服务通信模块技术方案书

1. 总体架构设计

2. 技术栈说明

3. 详细设计方案

3.1 gRPC接口定义 (helloworld.proto)

syntax = "proto3";

option java_multiple_files = true;
option java_package = "com.example.grpc";
option java_outer_classname = "HelloWorldProto";

service Greeter {
  rpc SayHello (HelloRequest) returns (HelloReply) {}
}

message HelloRequest {
  string name = 1;
}

message HelloReply {
  string message = 1;
}

3.2 服务端实现

3.2.1 核心组件

3.2.2 限流配置

3.3 客户端实现

3.3.1 连接管理策略

4. 代码实现

4.1 依赖配置 (pom.xml)

<dependencies>
    <!-- gRPC -->
    <dependency>
        <groupId>io.grpc</groupId>
        <artifactId>grpc-netty</artifactId>
        <version>1.58.0</version>
    </dependency>
    <dependency>
        <groupId>io.grpc</groupId>
        <artifactId>grpc-protobuf</artifactId>
        <version>1.58.0</version>
    </dependency>
    <dependency>
        <groupId>io.grpc</groupId>
        <artifactId>grpc-stub</artifactId>
        <version>1.58.0</version>
    </dependency>
    
    <!-- Sentinel -->
    <dependency>
        <groupId>com.alibaba.csp</groupId>
        <artifactId>sentinel-core</artifactId>
        <version>1.8.7</version>
    </dependency>
    <dependency>
        <groupId>com.alibaba.csp</groupId>
        <artifactId>sentinel-grpc</artifactId>
        <version>1.8.7</version>
    </dependency>
    
    <!-- 连接池 -->
    <dependency>
        <groupId>org.apache.commons</groupId>
        <artifactId>commons-pool2</artifactId>
        <version>2.11.1</version>
    </dependency>
</dependencies>

4.2 服务端实现

4.2.1 gRPC服务实现 (HelloServiceImpl.java)

public class HelloServiceImpl extends GreeterGrpc.GreeterImplBase {
    
    private static final Logger logger = LoggerFactory.getLogger(HelloServiceImpl.class);
    
    @Override
    public void sayHello(HelloRequest request, StreamObserver<HelloReply> responseObserver) {
        String name = request.getName();
        String message = "Hello, " + name + "!";
        
        HelloReply reply = HelloReply.newBuilder().setMessage(message).build();
        responseObserver.onNext(reply);
        responseObserver.onCompleted();
        
        logger.info("Processed request for: {}", name);
    }
}

4.2.2 Sentinel拦截器 (SentinelInterceptor.java)

public class SentinelInterceptor implements ServerInterceptor {
    
    private static final String RESOURCE_NAME = "grpc_service:SayHello";
    
    static {
        // 初始化限流规则：QPS=2
        FlowRule rule = new FlowRule();
        rule.setResource(RESOURCE_NAME);
        rule.setGrade(RuleConstant.FLOW_GRADE_QPS);
        rule.setCount(2);
        rule.setControlBehavior(RuleConstant.CONTROL_BEHAVIOR_DEFAULT);
        FlowRuleManager.loadRules(Collections.singletonList(rule));
    }

    @Override
    public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
            ServerCall<ReqT, RespT> call,
            Metadata headers,
            ServerCallHandler<ReqT, RespT> next) {
        
        // 资源名称根据方法名动态生成
        String resourceName = RESOURCE_NAME + ":" + call.getMethodDescriptor().getFullMethodName();
        
        Entry entry = null;
        try {
            entry = SphU.entry(resourceName, EntryType.IN);
            return next.startCall(call, headers);
        } catch (BlockException e) {
            // 限流处理
            call.close(Status.RESOURCE_EXHAUSTED.withDescription("Request blocked by Sentinel"), new Metadata());
            return new ServerCall.Listener<>() {};
        } finally {
            if (entry != null) {
                entry.exit();
            }
        }
    }
}

4.2.3 gRPC服务启动器 (GrpcServer.java)

public class GrpcServer {
    private Server server;
    
    public void start() throws IOException {
        int port = 50051;
        server = ServerBuilder.forPort(port)
                .addService(new HelloServiceImpl())
                .intercept(new SentinelInterceptor()) // 添加Sentinel拦截器
                .build()
                .start();
        
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            GrpcServer.this.stop();
        }));
    }
    
    public void stop() {
        if (server != null) {
            server.shutdown();
        }
    }
    
    public void blockUntilShutdown() throws InterruptedException {
        if (server != null) {
            server.awaitTermination();
        }
    }
}

4.3 客户端实现

4.3.1 连接池管理 (ChannelPoolFactory.java)

public class ChannelPoolFactory {
    private static final GenericObjectPool<ManagedChannel> channelPool;
    
    static {
        PooledObjectFactory<ManagedChannel> factory = new BasePooledObjectFactory<>() {
            @Override
            public ManagedChannel create() {
                return ManagedChannelBuilder.forAddress("localhost", 50051)
                        .usePlaintext()
                        .idleTimeout(30, TimeUnit.MINUTES) // 30分钟空闲超时
                        .keepAliveTime(60, TimeUnit.SECONDS) // 60秒心跳
                        .build();
            }
            
            @Override
            public PooledObject<ManagedChannel> wrap(ManagedChannel channel) {
                return new DefaultPooledObject<>(channel);
            }
            
            @Override
            public void destroyObject(PooledObject<ManagedChannel> p) {
                p.getObject().shutdown();
            }
        };
        
        GenericObjectPoolConfig<ManagedChannel> config = new GenericObjectPoolConfig<>();
        config.setMaxTotal(5); // 最大连接数
        config.setMinIdle(1);  // 最小空闲连接
        config.setMaxWaitMillis(3000); // 获取连接超时时间
        
        channelPool = new GenericObjectPool<>(factory, config);
    }
    
    public static ManagedChannel getChannel() throws Exception {
        return channelPool.borrowObject();
    }
    
    public static void returnChannel(ManagedChannel channel) {
        channelPool.returnObject(channel);
    }
}

4.3.2 客户端轮询逻辑 (GrpcClient.java)

public class GrpcClient {
    private static final Random random = new Random();
    private static final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
    
    public static void main(String[] args) {
        // 启动10个客户端线程模拟并发
        for (int i = 0; i < 10; i++) {
            scheduler.scheduleAtFixedRate(() -> makeRequest(), 
                0, 500 + random.nextInt(1500), TimeUnit.MILLISECONDS);
        }
    }
    
    private static void makeRequest() {
        ManagedChannel channel = null;
        try {
            channel = ChannelPoolFactory.getChannel();
            GreeterGrpc.GreeterBlockingStub stub = GreeterGrpc.newBlockingStub(channel);
            
            HelloRequest request = HelloRequest.newBuilder()
                .setName("Client-" + Thread.currentThread().getId())
                .build();
            
            try {
                HelloReply response = stub.sayHello(request);
                System.out.printf("[%s] Received: %s%n", 
                    Thread.currentThread().getName(), response.getMessage());
            } catch (StatusRuntimeException e) {
                if (e.getStatus().getCode() == Status.Code.RESOURCE_EXHAUSTED) {
                    System.err.printf("[%s] Request blocked by rate limiting%n", 
                        Thread.currentThread().getName());
                } else {
                    e.printStackTrace();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (channel != null) {
                ChannelPoolFactory.returnChannel(channel);
            }
        }
    }
}

5. 性能优化策略

5.1 连接管理优化

5.2 Sentinel高级配置

// 添加热点参数限流
ParamFlowRule rule = new ParamFlowRule(RESOURCE_NAME)
    .setParamIdx(0) // 第一个参数
    .setCount(5)
    .setGrade(RuleConstant.FLOW_GRADE_QPS)
    .setDurationInSec(1)
    .setParamFlowItemList(Collections.singletonList(
        new ParamFlowItem().setObject("highPriority")
            .setClassType(String.class.getName())
            .setCount(10)
    ));
ParamFlowRuleManager.loadRules(Collections.singletonList(rule));

5.3 监控与告警

监控指标：

1. 请求QPS与响应时间
2. 限流拒绝次数
3. 连接池使用率
4. 线程池活跃度

6. 部署方案

6.1 容器化部署 (Dockerfile)

FROM openjdk:17-jdk-slim

WORKDIR /app

COPY target/grpc-service.jar /app/app.jar

EXPOSE 50051

ENTRYPOINT ["java", "-jar", "app.jar"]

6.2 Kubernetes部署 (deployment.yaml)

apiVersion: apps/v1
kind: Deployment
metadata:
  name: grpc-service
spec:
  replicas: 3
  selector:
    matchLabels:
      app: grpc-service
  template:
    metadata:
      labels:
        app: grpc-service
    spec:
      containers:
      - name: grpc-service
        image: registry.example.com/grpc-service:1.0.0
        ports:
        - containerPort: 50051
        resources:
          limits:
            memory: "512Mi"
            cpu: "500m"
---
apiVersion: v1
kind: Service
metadata:
  name: grpc-service
spec:
  selector:
    app: grpc-service
  ports:
    - protocol: TCP
      port: 50051
      targetPort: 50051

7. 测试方案

7.1 性能测试脚本 (test.sh)

#!/bin/bash

# 启动服务端
java -jar grpc-server.jar &

# 等待服务启动
sleep 5

# 启动客户端测试
for i in {1..10}
do
   java -jar grpc-client.jar > client-$i.log &
done

# 监控限流情况
watch -n 1 "grep 'blocked' *.log | wc -l"

7.2 测试结果验证

8. 项目优势总结

1. 高性能通信：基于gRPC HTTP/2协议，支持多路复用和头部压缩
2. 精准流量控制：Sentinel实现毫秒级QPS限流
3. 资源高效利用：连接池管理减少TCP握手开销
4. 弹性扩展：无状态设计支持水平扩展
5. 生产就绪：集成健康检查、指标监控等生产级特性

部署说明：项目启动顺序为：1. 启动gRPC服务端 2. 启动gRPC客户端。Sentinel限流规则会在服务端启动时自动初始化。