当前位置：首页 > news >正文

grpc 和限流Sentinel

news 2025/7/4 9:25:46

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

1. 总体架构设计

2. 技术栈说明

组件	版本	功能
gRPC	1.58.0	高性能RPC框架
Protocol Buffers	3.24.4	接口定义与序列化
Sentinel	1.8.7	流量控制与熔断降级
Netty	4.1.100.Final	网络通信基础
Spring Boot	3.1.5	应用框架

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 限流配置

参数	值	说明
资源名	`grpc_service:SayHello`	Sentinel资源标识
阈值类型	QPS	每秒请求数
单机阈值	2	每秒最大请求数
流控效果	直接拒绝	超限直接返回错误

3.3 客户端实现

3.3.1 连接管理策略

参数	值	说明
连接池大小	5	最大连接数
空闲超时	30分钟	自动关闭空闲连接
心跳间隔	60秒	保持连接活跃

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);@Overridepublic 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=2FlowRule 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));}@Overridepublic <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<>() {@Overridepublic ManagedChannel create() {return ManagedChannelBuilder.forAddress("localhost", 50051).usePlaintext().idleTimeout(30, TimeUnit.MINUTES) // 30分钟空闲超时.keepAliveTime(60, TimeUnit.SECONDS) // 60秒心跳.build();}@Overridepublic PooledObject<ManagedChannel> wrap(ManagedChannel channel) {return new DefaultPooledObject<>(channel);}@Overridepublic 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 连接管理优化

策略	实现方式	效果
连接预热	启动时创建最小空闲连接	避免首次请求延迟
动态扩容	监控连接等待队列	自动增加连接池大小
健康检查	定期ping空闲连接	及时发现失效连接

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 监控与告警

监控指标：

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

6. 部署方案

6.1 容器化部署 (`Dockerfile`)

FROM openjdk:17-jdk-slimWORKDIR /appCOPY target/grpc-service.jar /app/app.jarEXPOSE 50051ENTRYPOINT ["java", "-jar", "app.jar"]

6.2 Kubernetes部署 (`deployment.yaml`)

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

7. 测试方案

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

#!/bin/bash# 启动服务端
java -jar grpc-server.jar &# 等待服务启动
sleep 5# 启动客户端测试
for i in {1..10}
dojava -jar grpc-client.jar > client-$i.log &
done# 监控限流情况
watch -n 1 "grep 'blocked' *.log | wc -l"

7.2 测试结果验证

测试场景	预期结果	验证方法
正常请求(QPS<2)	全部成功	响应成功率100%
限流触发(QPS>2)	部分拒绝	错误日志包含"blocked"
长连接保持	连接复用	连接创建日志次数<请求次数
高并发压力	服务稳定	CPU/内存波动在安全范围