gRPC三种Java客户端性能测试实践

Posted 2022-07-09 程序员二黑

本篇文章只做性能测试实践，不会测试各类状况下极限性能，所以硬件配置和软件参数就不单独分享了。

服务端

依旧采用了fun_grpc项目的SDK内容。服务端代码如下：

package com.funtester.grpc;

import com.funtester.frame.execute.ThreadPoolUtil;
import io.grpc.Server;
import io.grpc.ServerBuilder;

import java.io.IOException;
import java.util.concurrent.ThreadPoolExecutor;

public class Service 

    public static void main(String[] args) throws IOException, InterruptedException 
        ThreadPoolExecutor pool = ThreadPoolUtil.createFixedPool(10, "gRPC");
        Server server = ServerBuilder
                .forPort(12345)
                .executor(pool)
                .addService(new HelloServiceImpl())
                .build();

        server.start();
        server.awaitTermination();
    

实际业务处理类：

package com.funtester.grpc;

import com.funtester.frame.SourceCode;
import com.funtester.fungrpc.HelloRequest;
import com.funtester.fungrpc.HelloResponse;
import com.funtester.fungrpc.HelloServiceGrpc;
import com.funtester.utils.Time;
import io.grpc.stub.StreamObserver;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

public class HelloServiceImpl extends HelloServiceGrpc.HelloServiceImplBase 

    private static final Logger logger = LogManager.getLogger(HelloServiceImpl.class);

    @Override
    public void executeHi(HelloRequest request, StreamObserver<HelloResponse> responseObserver) 
        HelloResponse response = HelloResponse.newBuilder()
                .setMsg("你好 " + request.getName()+ Time.getDate())
                .build();
        SourceCode.sleep(1.0);
        logger.info("用户来了",request.getName());
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    

业务上休眠了1s，然后返回响应内容。

客户端

客户端实际使用相对简单，这里就不再分享了，有兴趣的可以文末加群讨论

静态模型

首先分享一下静态模型的内容，所谓静态内容指的是用例执行之前就设定好了执行的整个过程，用例执行过程除了终止以外没有其他干预措施。

线程模型

下面是基于静态线程模型的性能测试用例：

package com.funtest.grpc

import com.funtester.base.constaint.FixedThread
import com.funtester.frame.SourceCode
import com.funtester.frame.execute.Concurrent
import com.funtester.fungrpc.HelloRequest
import com.funtester.fungrpc.HelloServiceGrpc
import io.grpc.ManagedChannel
import io.grpc.ManagedChannelBuilder

class FixedThreadModel extends SourceCode 

    static int times

    static HelloServiceGrpc.HelloServiceBlockingStub helloServiceBlockingStub

    static HelloRequest requst

    public static void main(String[] args) 
        ManagedChannel managedChannel = ManagedChannelBuilder.forAddress("localhost", 12345)
                .usePlaintext().build()

        helloServiceBlockingStub = HelloServiceGrpc.newBlockingStub(managedChannel).withCompression("gzip")
        requst = HelloRequest.newBuilder()
                .setName("FunTester")
                .build()
        RUNUP_TIME = 0
        times = 2000
        new Concurrent(new FunTester(), 10, "静态线程模型").start()

        managedChannel.shutdown()

    

    private static class FunTester extends FixedThread 


        FunTester() 
            super(null, times, true)
        

        @Override
        protected void doing() throws Exception 
            helloServiceBlockingStub.executeHi(requst)
        

        @Override
        FunTester clone() 
            return new FunTester()
        
    

QPS模型

下面是基于静态QPS模型的压测用例。

package com.funtest.grpc

import com.funtester.base.event.FunCount
import com.funtester.frame.SourceCode
import com.funtester.frame.execute.FunEventConcurrent
import com.funtester.fungrpc.HelloRequest
import com.funtester.fungrpc.HelloServiceGrpc
import io.grpc.ManagedChannel
import io.grpc.ManagedChannelBuilder

class FixedQpsModel extends SourceCode 

    static HelloServiceGrpc.HelloServiceBlockingStub helloServiceBlockingStub

    static HelloRequest requst

    public static void main(String[] args) 
        ManagedChannel managedChannel = ManagedChannelBuilder.forAddress("localhost", 12345)
                .usePlaintext().build()

        helloServiceBlockingStub = HelloServiceGrpc.newBlockingStub(managedChannel).withCompression("gzip")
        requst = HelloRequest.newBuilder()
                .setName("FunTester")
                .build()
        def count = new FunCount(1, 1, 2, 1000, 10, "静态QPS模型")

        def test= 
            helloServiceBlockingStub.executeHi(requst)
        
        new FunEventConcurrent(test,count).start()
        managedChannel.shutdown()

    

以上是两个常用的静态模型的演示，还有其他的动态模型这里就不演示了。

动态模型

下面到了喜闻乐见的动态模型的part，动态模型值得是用例执行时都是以固定的最小压力值（通常为1个QPS或者1个线程）启动，然后再用例执行过程中不断调整（调整步长、增减）用例的压力。

动态线程模型

由于动态模型是不限制用例运行时间，所以取消了关闭channel的方法。

package com.funtest.grpc

import com.funtester.base.constaint.FunThread
import com.funtester.frame.SourceCode
import com.funtester.frame.execute.FunConcurrent
import com.funtester.fungrpc.HelloRequest
import com.funtester.fungrpc.HelloServiceGrpc
import io.grpc.ManagedChannel
import io.grpc.ManagedChannelBuilder

import java.util.concurrent.atomic.AtomicInteger

class FunThreadModel extends SourceCode 

    static int times

    static HelloServiceGrpc.HelloServiceBlockingStub helloServiceBlockingStub

    static HelloRequest requst

    static AtomicInteger index = new AtomicInteger(0)

    static def desc = "动态线程模型"

    public static void main(String[] args) 
        ManagedChannel managedChannel = ManagedChannelBuilder.forAddress("localhost", 12345)
                .usePlaintext().build()

        helloServiceBlockingStub = HelloServiceGrpc.newBlockingStub(managedChannel).withCompression("gzip")
        requst = HelloRequest.newBuilder()
                .setName("FunTester")
                .build()
        new FunConcurrent(new FunTester()).start()
    

    private static class FunTester extends FunThread 


        FunTester() 
            super(null, desc + index.getAndIncrement())
        

        @Override
        protected void doing() throws Exception 
            helloServiceBlockingStub.executeHi(requst)
        

        @Override
        FunTester clone() 
            return new FunTester()
        
    

动态QPS模型

动态QPS模型是我现在最常用的模型，优势多多，除了某些强用户绑定需求外，动态QPS模型都是第一选择。

package com.funtest.grpc


import com.funtester.frame.SourceCode
import com.funtester.frame.execute.FunQpsConcurrent
import com.funtester.fungrpc.HelloRequest
import com.funtester.fungrpc.HelloServiceGrpc
import io.grpc.ManagedChannel
import io.grpc.ManagedChannelBuilder

class FunQpsModel extends SourceCode 

    static HelloServiceGrpc.HelloServiceBlockingStub helloServiceBlockingStub

    static HelloRequest requst

    public static void main(String[] args) 
        ManagedChannel managedChannel = ManagedChannelBuilder.forAddress("localhost", 12345)
                .usePlaintext().build()

        helloServiceBlockingStub = HelloServiceGrpc.newBlockingStub(managedChannel).withCompression("gzip")
        requst = HelloRequest.newBuilder()
                .setName("FunTester")
                .build()
        def test= 
            helloServiceBlockingStub.executeHi(requst)
        
        new FunQpsConcurrent(test).start()

    

以上就是常用的gRPC阻塞客户端四种模型的性能测试全部内容了，欢迎关注我。

学习资源分享

最后感谢每一个认真阅读我文章的人，看着粉丝一路的上涨和关注，礼尚往来总是要有的，虽然不是什么很值钱的东西，如果你用得到的话可以直接拿走

这些资料，对于想进阶【自动化测试】的朋友来说应该是最全面最完整的备战仓库，这个仓库也陪伴我走过了最艰难的路程，希望也能帮助到你！凡事要趁早，特别是技术行业，一定要提升技术功底。希望对大家有所帮助…….

加入下方我的交流群免费获取！