性能工具之Jmeter压测Thrift RPC服务

2024-01-18 18:30:16

概述

Thrift是一个可互操作和可伸缩服务的框架，用来进行可扩展且跨语言的服务的开发。它结合了功能强大的软件堆栈和代码生成引擎，以构建在 C++, Java, Python, PHP, Ruby, Erlang, Perl, Haskell, C#, Cocoa, JavaScript, Node.js, Smalltalk, and OCaml 等等编程语言间无缝结合的、高效的服务。

Thrift最初由facebook开发，07年四月开放源码，08年5月进入apache孵化器。thrift允许你定义一个简单的定义文件中的数据类型和服务接口（IDL）。以作为输入文件，编译器生成代码用来方便地生成RPC客户端和服务器通信的无缝跨编程语言。

其传输数据采用二进制格式，相对于XML和JSON等序列化方式体积更小，对于高并发、大数据量和多语言的环境更有优势。 Thrift它含有三个主要的组件：protocol，transport和server，其中，protocol定义了消息是怎样序列化的，transport定义了消息是怎样在客户端和服务器端之间通信的，server用于从transport接收序列化的消息，根据protocol反序列化之，调用用户定义的消息处理器，并序列化消息处理器的响应，然后再将它们写回transport。

官网地址：thrift.apache.org

基本概念

架构图

堆栈的顶部是从Thrift定义文件生成的代码。Thrift 服务生成的客户端和处理器代码。这些由图中的棕色框表示。红色框为发送的数据结构（内置类型除外）也会生成代码。协议和传输是Thrift运行时库的一部分。因此使用Thrift可以定义服务，并且可以*更改协议和传输，而无需重新生成代码。 Thrift还包括一个服务器基础结构，用于将协议和传输绑定在一起。有可用的阻塞，非阻塞，单线程和多线程服务器。堆栈的“底层I / O”部分根据所开发语言而有所不同。对于Java和Python网络I / O，Thrift库利用内置库，而C ++实现使用自己的自定义实现。

数据类型：

基本类型：

bool：布尔值，true 或 false，对应 Java 的 boolean
byte：8 位有符号整数，对应 Java 的 byte
i16：16 位有符号整数，对应 Java 的 short
i32：32 位有符号整数，对应 Java 的 int
i64：64 位有符号整数，对应 Java 的 long
double：64 位浮点数，对应 Java 的 double
string：未知编码文本或二进制字符串，对应 Java 的 String

结构体类型：

struct：定义公共的对象，类似于 C 语言中的结构体定义，在 Java 中是一个 JavaBean

集合类型：

list：对应 Java 的 ArrayList
set：对应 Java 的 HashSet
map：对应 Java 的 HashMap

异常类型：

exception：对应 Java 的 Exception

服务类型：

service：对应服务的类

数据传输层Transport

TSocket —— 使用阻塞式 I/O 进行传输，是最常见的模式
TFramedTransport —— 使用非阻塞方式，按块的大小进行传输，类似于 Java 中的 NIO，若使用 TFramedTransport 传输层，其服务器必须修改为非阻塞的服务类型
TNonblockingTransport —— 使用非阻塞方式，用于构建异步客户端

数据传输协议Protocol

Thrift 可以让用户选择客户端与服务端之间传输通信协议的类别，在传输协议上总体划分为文本 (text) 和二进制 (binary) 传输协议，为节约带宽，提高传输效率，一般情况下使用二进制类型的传输协议为多数，有时还会使用基于文本类型的协议，这需要根据项目 / 产品中的实际需求。

常用协议有以下几种：

TBinaryProtocol : 二进制格式.
TCompactProtocol : 高效率的、密集的二进制压缩格式
TJSONProtocol : JSON格式
TSimpleJSONProtocol : 提供JSON只写协议, 生成的文件很容易通过脚本语言解析

注意：客户端和服务端的协议要一致。

服务器类型Server

TSimpleServer ——单线程服务器端使用标准的阻塞式 I/O，一般用于测试。
TThreadPoolServer —— 多线程服务器端使用标准的阻塞式 I/O，预先创建一组线程处理请求。
TNonblockingServer —— 多线程服务器端使用非阻塞式 I/O，服务端和客户端需要指定 TFramedTransport 数据传输的方式。
THsHaServer —— 半同步半异步的服务端模型，需要指定为： TFramedTransport 数据传输的方式。它使用一个单独的线程来处理网络I/O，一个独立的worker线程池来处理消息。这样，只要有空闲的worker线程，消息就会被立即处理，因此多条消息能被并行处理。
TThreadedSelectorServer —— TThreadedSelectorServer允许你用多个线程来处理网络I/O。它维护了两个线程池，一个用来处理网络I/O，另一个用来进行请求的处理。当网络I/O是瓶颈的时候，TThreadedSelectorServer比THsHaServer的表现要好。

实现逻辑

服务端

实现服务处理接口?impl

创建TProcessor 创建TServerTransport 创建TProtocol 创建TServer 启动Server

客户端

创建Transport 创建TProtocol 基于TTransport和TProtocol创建 Client 调用Client的相应方法

ThriftServerDemo实例

新建?Maven项目，并且添加?thrift依赖包

?<dependencies>
? ? ? ?<dependency>
? ? ? ? ? ?<groupId>org.apache.thrift</groupId>
? ? ? ? ? ?<artifactId>libthrift</artifactId>
? ? ? ? ? ?<version>0.9.3</version>
? ? ? ?</dependency>
? ? ? ?<dependency>
? ? ? ? ? ?<groupId>org.slf4j</groupId>
? ? ? ? ? ?<artifactId>slf4j-log4j12</artifactId>
? ? ? ? ? ?<version>1.7.12</version>
? ? ? ?</dependency>
? ? ? ?<dependency>
? ? ? ? ? ?<groupId>org.apache.logging.log4j</groupId>
? ? ? ? ? ?<artifactId>log4j-api</artifactId>
? ? ? ? ? ?<version>2.7</version>
? ? ? ?</dependency>
? ? ? ?<dependency>
? ? ? ? ? ?<groupId>org.apache.logging.log4j</groupId>
? ? ? ? ? ?<artifactId>log4j-core</artifactId>
? ? ? ? ? ?<version>2.7</version>
? ? ? ?</dependency>
? ?</dependencies>
? ?<build>
? ? ? ?<plugins>
? ? ? ? ? ?<plugin>
? ? ? ? ? ? ? ?<groupId>org.apache.maven.plugins</groupId>
? ? ? ? ? ? ? ?<artifactId>maven-compiler-plugin</artifactId>
? ? ? ? ? ? ? ?<version>3.3</version>
? ? ? ? ? ? ? ?<configuration>
? ? ? ? ? ? ? ? ? ?<source>1.8</source>
? ? ? ? ? ? ? ? ? ?<target>1.8</target>
? ? ? ? ? ? ? ? ? ?<encoding>utf-8</encoding>
? ? ? ? ? ? ? ?</configuration>
? ? ? ? ? ?</plugin>
? ? ? ?</plugins>
? ?</build>

编写?IDL接口并生成接口文件

namespace java thrift.service
// 计算类型 - 仅限整数四则运算
enum ComputeType {
? ?ADD = 0;
? ?SUB = 1;
? ?MUL = 2;
? ?DIV = 3;
}
// 服务请求
struct ComputeRequest {
? ?1:required i64 x;
? ?2:required i64 y;
? ?3:required ComputeType computeType;
}
// 服务响应
struct ComputeResponse {
? ?1:required i32 errorNo;
? ?2:optional string errorMsg;
? ?3:required i64 computeRet;
}
service ComputeServer {
? ?ComputeResponse getComputeResult(1:ComputeRequest request);
}

执行编译命令：

thrift-0.11.0.exe -r -gen java computeServer.thrift

拷贝生成的?Service类文件到?IDEA?

服务端接口实现

public class ThriftTestImpl implements ComputeServer.Iface {
? ?private static final Logger logger = LogManager.getLogger(ThriftTestImpl.class);
? ?public ComputeResponse getComputeResult(ComputeRequest request) {
? ? ? ?ComputeType computeType = request.getComputeType();
? ? ? ?long x = request.getX();
? ? ? ?long y = request.getY();
? ? ? ?logger.info("get compute result begin. [x:{}] [y:{}] [type:{}]", x, y, computeType.toString());
? ? ? ?long begin = System.currentTimeMillis();
? ? ? ?ComputeResponse response = new ComputeResponse();
? ? ? ?response.setErrorNo(0);
? ? ? ?try {
? ? ? ? ? ?long ret;
? ? ? ? ? ?if (computeType == ComputeType.ADD) {
? ? ? ? ? ? ? ?ret = add(x, y);
? ? ? ? ? ? ? ?response.setComputeRet(ret);
? ? ? ? ? ?} else if (computeType == ComputeType.SUB) {
? ? ? ? ? ? ? ?ret = sub(x, y);
? ? ? ? ? ? ? ?response.setComputeRet(ret);
? ? ? ? ? ?} else if (computeType == ComputeType.MUL) {
? ? ? ? ? ? ? ?ret = mul(x, y);
? ? ? ? ? ? ? ?response.setComputeRet(ret);
? ? ? ? ? ?} else {
? ? ? ? ? ? ? ?ret = div(x, y);
? ? ? ? ? ? ? ?response.setComputeRet(ret);
? ? ? ? ? ?}
? ? ? ?} catch (Exception e) {
? ? ? ? ? ?response.setErrorNo(1001);
? ? ? ? ? ?response.setErrorMsg(e.getMessage());
? ? ? ? ? ?logger.error("exception:", e);
? ? ? ?}
? ? ? ?long end = System.currentTimeMillis();
? ? ? ?logger.info("get compute result end. [errno:{}] cost:[{}ms]", response.getErrorNo(), (end - begin));
? ? ? ?return response;
? ?}
? ?private long add(long x, long y) {
? ? ? ?return x + y;
? ?}
? ?private long sub(long x, long y) {
? ? ? ?return x - y;
? ?}
? ?private long mul(long x, long y) {
? ? ? ?return x * y;
? ?}
? ?private long div(long x, long y) {
? ? ? ?return x / y;
? ?}
}

服务端实现

public class ServerMain {
? ?private static final Logger logger = LogManager.getLogger(ServerMain.class);
? ?public static void main(String[] args) {
? ? ? ?try {
? ? ? ? ? ?//实现服务处理接口impl
? ? ? ? ? ?ThriftTestImpl workImpl = new ThriftTestImpl();
? ? ? ? ? ?//创建TProcessor
? ? ? ? ? ?TProcessor tProcessor = new ComputeServer.Processor<ComputeServer.Iface>(workImpl);
? ? ? ? ? ?//创建TServerTransport,非阻塞式 I/O，服务端和客户端需要指定 TFramedTransport 数据传输的方式
? ? ? ? ? ?final TNonblockingServerTransport transport = new TNonblockingServerSocket(9999);
? ? ? ? ? ?//创建TProtocol
? ? ? ? ? ?TThreadedSelectorServer.Args ttpsArgs = new TThreadedSelectorServer.Args(transport);
? ? ? ? ? ?ttpsArgs.transportFactory(new TFramedTransport.Factory());
? ? ? ? ? ?//二进制格式反序列化
? ? ? ? ? ?ttpsArgs.protocolFactory(new TBinaryProtocol.Factory());
? ? ? ? ? ?ttpsArgs.processor(tProcessor);
? ? ? ? ? ?ttpsArgs.selectorThreads(16);
? ? ? ? ? ?ttpsArgs.workerThreads(32);
? ? ? ? ? ?logger.info("compute service server on port :" + 9999);
? ? ? ? ? ?//创建TServer
? ? ? ? ? ?TServer server = new TThreadedSelectorServer(ttpsArgs);
? ? ? ? ? ?//启动Server
? ? ? ? ? ?server.serve();
? ? ? ?} catch (Exception e) {
? ? ? ? ? ?logger.error(e);
? ? ? ?}
? ?}
}

服务端整体代码结构?

log4j2.xml配置文件

<?xml version="1.0" encoding="UTF-8"?>



<configuration status="INFO" monitorInterval="30">
? ?
? ?<appenders>
? ? ? ?
? ? ? ?<console name="Console" target="SYSTEM_OUT">
? ? ? ? ? ?
? ? ? ? ? ?<PatternLayout pattern="%highlight{[ %p ] [%-d{yyyy-MM-dd HH:mm:ss}] [%l] %m%n}"/>
? ? ? ?</console>
? ? ? ?<RollingFile name="RollingFileInfo" fileName="log/log.log" filePattern="log/log.log.%d{yyyy-MM-dd}">
? ? ? ? ? ?
? ? ? ? ? ?<ThresholdFilter level="info" onMatch="ACCEPT" onMismatch="DENY"/>
? ? ? ? ? ?<PatternLayout pattern="[ %p ] [%-d{yyyy-MM-dd HH:mm:ss}] [ LOGID:%X{logid} ] [%l] %m%n"/>
? ? ? ? ? ?<Policies>
? ? ? ? ? ? ? ?<TimeBasedTriggeringPolicy modulate="true" interval="1"/>
? ? ? ? ? ? ? ?<SizeBasedTriggeringPolicy/>
? ? ? ? ? ?</Policies>
? ? ? ?</RollingFile>
? ? ? ?<RollingFile name="RollingFileError" fileName="log/error.log" filePattern="log/error.log.%d{yyyy-MM-dd}">
? ? ? ? ? ?
? ? ? ? ? ?<Filters>
? ? ? ? ? ? ? ?<ThresholdFilter level="warn" onMatch="ACCEPT" onMismatch="DENY" />
? ? ? ? ? ?</Filters>
? ? ? ? ? ?<PatternLayout pattern="[ %p ] %-d{yyyy-MM-dd HH:mm:ss} [ %t:%r ] [%l] %m%n"/>
? ? ? ? ? ?<Policies>
? ? ? ? ? ? ? ?<TimeBasedTriggeringPolicy modulate="true" interval="1"/>
? ? ? ? ? ? ? ?<SizeBasedTriggeringPolicy/>
? ? ? ? ? ?</Policies>
? ? ? ?</RollingFile>
? ?</appenders>
? ?
? ?<loggers>
? ? ? ?
? ? ? ?<logger name="org.springframework" level="INFO"></logger>
? ? ? ?<logger name="org.mybatis" level="INFO"></logger>
? ? ? ?<root level="all">
? ? ? ? ? ?<appender-ref ref="Console"/>
? ? ? ? ? ?<appender-ref ref="RollingFileInfo"/>
? ? ? ? ? ?<appender-ref ref="RollingFileError"/>
? ? ? ?</root>
? ?</loggers>
</configuration>

Jmeter测试类编写

利用JMeter调用Java测试类去调用对应的后台服务，并记住每次调用并获取反馈值的RT,ERR%，只需要按照单线程的方式去实现测试业务，也无需添加各种埋点收集数据?

新建一个?JavaMaven工程,添加?JMeter及?thrift依赖包

<dependencies>
? ? ? ?<dependency>
? ? ? ? ? ?<groupId>org.apache.jmeter</groupId>
? ? ? ? ? ?<artifactId>ApacheJMeter_core</artifactId>
? ? ? ? ? ?<version>4.0</version>
? ? ? ?</dependency>
? ? ? ?<dependency>
? ? ? ? ? ?<groupId>org.apache.jmeter</groupId>
? ? ? ? ? ?<artifactId>ApacheJMeter_java</artifactId>
? ? ? ? ? ?<version>4.0</version>
? ? ? ?</dependency>
? ? ? ?<dependency>
? ? ? ? ? ?<groupId>org.apache.thrift</groupId>
? ? ? ? ? ?<artifactId>libthrift</artifactId>
? ? ? ? ? ?<version>0.11.0</version>
? ? ? ?</dependency>
? ? ? ?<dependency>
? ? ? ? ? ?<groupId>org.apache.logging.log4j</groupId>
? ? ? ? ? ?<artifactId>log4j-api</artifactId>
? ? ? ? ? ?<version>2.11.1</version>
? ? ? ?</dependency>
? ? ? ?<dependency>
? ? ? ? ? ?<groupId>org.apache.logging.log4j</groupId>
? ? ? ? ? ?<artifactId>log4j-core</artifactId>
? ? ? ? ? ?<version>2.11.1</version>
? ? ? ?</dependency>
? ? ? ?<dependency>
? ? ? ? ? ?<groupId>org.slf4j</groupId>
? ? ? ? ? ?<artifactId>slf4j-log4j12</artifactId>
? ? ? ? ? ?<version>1.7.25</version>
? ? ? ?</dependency>
? ?</dependencies>
? ?<build>
? ? ? ?<plugins>
? ? ? ? ? ?<plugin>
? ? ? ? ? ? ? ?<groupId>org.apache.maven.plugins</groupId>
? ? ? ? ? ? ? ?<artifactId>maven-compiler-plugin</artifactId>
? ? ? ? ? ? ? ?<version>3.7.0</version>
? ? ? ? ? ? ? ?<configuration>
? ? ? ? ? ? ? ? ? ?<source>1.8</source>
? ? ? ? ? ? ? ? ? ?<target>1.8</target>
? ? ? ? ? ? ? ? ? ?<encoding>utf-8</encoding>
? ? ? ? ? ? ? ?</configuration>
? ? ? ? ? ?</plugin>
? ? ? ?</plugins>
? ?</build>

ThriftClient测试类编写

public class ThriftClient {
? ?private ComputeServer.Client client = null;
? ?private TTransport tTransport = null;
? ?public ThriftClient(String ip,int port){
? ? ? ?try {
? ? ? ? ? ?TTransport tTransport = new TFramedTransport(new TSocket(ip,port));
? ? ? ? ? ?tTransport.open();
? ? ? ? ? ?TProtocol tProtocol = new TBinaryProtocol(tTransport);
? ? ? ? ? ?client = new ComputeServer.Client(tProtocol);
? ? ? ?} catch (TTransportException e) {
? ? ? ? ? ?e.printStackTrace();
? ? ? ?}
? ?}
? ?public ComputeResponse getResponse(ComputeRequest request){
? ? ? ?try {
? ? ? ? ? ?ComputeResponse response = client.getComputeResult(request);
? ? ? ? ? ?return response;
? ? ? ?} catch (TException e) {
? ? ? ? ? ?e.printStackTrace();
? ? ? ? ? ?return null;
? ? ? ?}
? ?}
? ?public void ?close(){
? ? ? ?if (tTransport != null && tTransport.isOpen()){
? ? ? ? ? ?tTransport.close();
? ? ? ?}
? ?}
}

注意：需要把编写?IDL接口文件拷贝到工程里

新建一个?JavaClass，如下例中的?TestThriftByJmeter，并继承?AbstractJavaSamplerClient。?AbstractJavaSamplerClient中默认实现了四个可以覆盖的方法，分别是?getDefaultParameters()，?setupTest()，?runTest()和?teardownTest()方法。

getDefaultParameters?方法主要用于设置传入界面的参数；
setupTest方法为初始化方法，用于初始化性能测试时的每个线程；
runTest方法为性能测试时的线程运行体；
teardownTest方法为测试结束方法，用于结束性能测试中的每个线程。

编写TestThriftByJmeter测试类

public class TestThriftByJmeter extends AbstractJavaSamplerClient {
? ?private ThriftClient client;
? ?private ComputeRequest request;
? ?private ComputeResponse response;
? ?//设置传入界面的参数
? ?@Override
? ?public Arguments getDefaultParameters(){
? ? ? ?Arguments arguments = new Arguments();
? ? ? ?arguments.addArgument("ip","172.16.14.251");
? ? ? ?arguments.addArgument("port","9999");
? ? ? ?arguments.addArgument("X","0");
? ? ? ?arguments.addArgument("Y","0");
? ? ? ?arguments.addArgument("type","0");
? ? ? ?return arguments;
? ?}
? ?//初始化方法
? ?@Override
? ?public void setupTest(JavaSamplerContext context){
? ? ? ?//获取Jmeter中设置的参数
? ? ? ?String ip = context.getParameter("ip");
? ? ? ?int port = context.getIntParameter("port");
? ? ? ?int x = context.getIntParameter("X");
? ? ? ?int y = context.getIntParameter("Y");
? ? ? ?ComputeType type = ComputeType.findByValue(context.getIntParameter("type"));
? ? ? ?//创建客户端
? ? ? ?client = new ThriftClient(ip,port);
? ? ? ?//设置request请求
? ? ? ?request = new ComputeRequest(x,y,type);
? ? ? ?super.setupTest(context);
? ?}
? ?//性能测试线程运行体
? ?@Override
? ?public SampleResult runTest(JavaSamplerContext context) {
? ? ? ?SampleResult result = new SampleResult();
? ? ? ?//开始统计响应时间标记
? ? ? ?result.sampleStart();
? ? ? ?try {
? ? ? ? ? ?long begin = System.currentTimeMillis();
? ? ? ? ? ?response = client.getResponse(request);
? ? ? ? ? ?long cost = (System.currentTimeMillis() - begin);
? ? ? ? ? ?//打印时间戳差值。Java请求响应时间
? ? ? ? ? ?System.out.println(response.toString()+" 总计花费：["+cost+"ms]");
? ? ? ? ? ?if (response == null){
? ? ? ? ? ? ? ?//设置测试结果为fasle
? ? ? ? ? ? ? ?result.setSuccessful(false);
? ? ? ? ? ? ? ?return result;
? ? ? ? ? ?}
? ? ? ? ? ?if (response.getErrorNo() == 0){
? ? ? ? ? ? ? ?//设置测试结果为true
? ? ? ? ? ? ? ?result.setSuccessful(true);
? ? ? ? ? ?}else{
? ? ? ? ? ? ? ?result.setSuccessful(false);
? ? ? ? ? ? ? ?result.setResponseMessage("ERROR");
? ? ? ? ? ?}
? ? ? ?}catch (Exception e){
? ? ? ? ? ?result.setSuccessful(false);
? ? ? ? ? ?result.setResponseMessage("ERROR");
? ? ? ? ? ?e.printStackTrace();
? ? ? ?}finally {
? ? ? ? ? ?//结束统计响应时间标记
? ? ? ? ? ?result.sampleEnd();
? ? ? ?}
? ? ? ?return result;
? ?}
? ?//测试结束方法
? ?public void tearDownTest(JavaSamplerContext context) {
? ? ? ?if (client != null) {
? ? ? ? ? ?client.close();
? ? ? ?}
? ? ? ?super.teardownTest(context);
? ?}
}