首页 > 代码库 > python grpc 应用
python grpc 应用
1、rpc介绍
2、grpc
3、基于grpc协议文件传输
4、基于grpc协议jmeter压测获取实时结果
5、基于grcp协议获取jmeter最终压测报告,并将报告保存至client端
6、压测中途停止jmeter
grpc server (jmeter server) 192.168.18.128 rpc client (本机)
1、rpc协议介绍
RPC(Remote Procedure Call Protocol)——远程过程调用协议,它是一种通过网络从远程计算机程序上请求服务,而不需要了解底层网络技术的协议。RPC协议假定某些传输协议的存在,如TCP或UDP,为通信程序之间携带信息数据。在OSI网络通信模型中,RPC跨越了传输层和应用层。RPC使得开发包括网络分布式多程序在内的应用程序更加容易。
RPC采用客户机/服务器模式。请求程序就是一个客户机,而服务提供程序就是一个服务器。首先,客户机调用进程发送一个有进程参数的调用信息到服务进程,然后等待应答信息。在服务器端,进程保持睡眠状态直到调用信息到达为止。当一个调用信息到达,服务器获得进程参数,计算结果,发送答复信息,然后等待下一个调用信息,最后,客户端调用进程接收答复信息,获得进程结果,然后调用执行继续进行。 -- 摘自 百度百科
2、grpc介绍
gRPC 是一款高性能、开源的 RPC 框架,产自 Google,基于 ProtoBuf
序列化协议进行开发,支持多种语言(Golang、Python、Java等),本篇只介绍 Python 的 gRPC 使用。因为 gRPC 对
HTTP/2 协议的支持使其在 Android、IOS 等客户端后端服务的开发领域具有良好的前景。gRPC
提供了一种简单的方法来定义服务,同时客户端可以充分利用 HTTP2 stream 的特性,从而有助于节省带宽、降低 TCP
的连接次数、节省CPU的使用等。
3、基于grpc协议文件传输
目标:将本地的文件(test.file)传输至rpc server (192.168.18.128)
python 版本 2.7.13
3.1、
gRPC 的安装:
[root@vm6 rpc]# pip2.7 install requests [root@vm6 rpc]# pip2.7 install grpcio
安装 ProtoBuf 相关的 python 依赖库:
[root@vm6 rpc]# pip2.7 install protobuf
安装 python grpc 的 protobuf 编译工具:
[root@vm6 rpc]# pip2.7 install grpcio-tools
定义 gRPC 接口:
[root@vm6 ~]# mkdir rpc rpc/__init__.py [root@vm6 ~]# cd rpc/ [root@vm6 rpc]# vim rpc.proto // grpc 版本 syntax = "proto3"; package rpc; //定义接口 service RPC { //相当于定义接口方法 rpc sendConfFile(Content) returns (Status) {} } //相当于定义类属性,此属性用于接受文本 message Content { string text = 1; } //此属性用于return 结果状态吗 message Status { int64 code = 1; }
编译 protobuf:
[root@vm6 rpc]# python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. ./rpc.proto
server 端代码:
# -*- encoding=utf-8 -*- from concurrent import futures import time import subprocess import codecs import sys import os import grpc import rpc_pb2 import rpc_pb2_grpc _ONE_DAY_IN_SECONDS = 60 * 60 * 24 #server端文件保存的位置 jmeter_config = os.path.join(os.getcwd(),r‘conf/config.jmx‘) class Performance(rpc_pb2_grpc.RPCServicer): def sendConfFile(self, content,context): ‘‘‘ 保存配置文件,如config.jmx ‘‘‘ text = content.text try: print jmeter_config conf_handle = codecs.open(jmeter_config,‘w‘,encoding=‘utf-8‘) conf_handle.write(text) return rpc_pb2.Status(code=0) except Exception,e: print e return rpc_pb2.Status(code=1) def serve(): server = grpc.server(futures.ThreadPoolExecutor(max_workers=10)) rpc_pb2_grpc.add_RPCServicer_to_server(Performance(),server) server.add_insecure_port(‘[::]:50051‘) server.start() try: while True: time.sleep(_ONE_DAY_IN_SECONDS) except KeyboardInterrupt: server.stop(0) if __name__ == ‘__main__‘: serve()
client 端代码:
# -*- encoding=utf-8 -*- from concurrent import futures import time import subprocess import codecs import sys import os import logging import json import grpc import rpc_pb2 import rpc_pb2_grpc logging.basicConfig(level=logging.DEBUG, format=‘%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s‘, datefmt=‘%a, %d %b %Y %H:%M:%S‘, filename=‘rpc.log‘, filemode=‘a+‘) rpc_server = r‘192.168.18.128‘ rpc_port = ‘50051‘ class performance(): ‘‘‘性能测试的客户端接口‘‘‘ def __init__(self,ip,port): ‘‘‘初始化,连接RPC服务‘‘‘ logging.info("performance_client init") conn = grpc.insecure_channel(ip + ‘:‘ + port) self.stub_client = rpc_pb2_grpc.RPCStub(channel=conn) def sendConfig(self,filename): file_handle = codecs.open(filename, ‘r‘, encoding=‘utf-8‘) content = file_handle.read() ‘‘‘向RPC server发送测试的配置文件‘‘‘ response = self.stub_client.sendConfFile(rpc_pb2.Content(text=content)) print response.code if __name__ == ‘__main__‘: client = performance(rpc_server,rpc_port) ##将本地的rpc_client.py文件当做测试文件发送到server端 client.sendConfig(‘rpc_client.py‘)
启动server端:
[root@vm6 rpc]# python rpc_server.py
启动client端:
D:\xisuo\rpc>python rpc_client.py 0
查看server端的文件:
[root@vm6 rpc]# more conf/config.jmx # -*- encoding=utf-8 -*- from concurrent import futures import time import subprocess import codecs import sys import os import logging 。。。略
4、基于grpc协议jmeter压测获取实时结果
4.1、server端部署jdk,jmeter
将jdk解压到/usr/local/ 配置环境变量
将jmeter解压到/usr/local
4.2、因为我们是有jmeter的no gui模式在Linux执行,故需要jmeter的jmx文件,我们可以在本地使用gui模式先生成jmx文件
最终结果文件为 test.jmx,内容如下:
<?xml version="1.0" encoding="UTF-8"?> <jmeterTestPlan version="1.2" properties="2.4" jmeter="2.9 r1437961"> <hashTree> <TestPlan guiclass="TestPlanGui" testclass="TestPlan" testname="测试计划" enabled="true"> <stringProp name="TestPlan.comments"></stringProp> <boolProp name="TestPlan.functional_mode">false</boolProp> <boolProp name="TestPlan.serialize_threadgroups">false</boolProp> <elementProp name="TestPlan.user_defined_variables" elementType="Arguments" guiclass="ArgumentsPanel" testclass="Arguments" testname="用户定义的变量" enabled="true"> <collectionProp name="Arguments.arguments"/> </elementProp> <stringProp name="TestPlan.user_define_classpath"></stringProp> </TestPlan> <hashTree> <ThreadGroup guiclass="ThreadGroupGui" testclass="ThreadGroup" testname="测试" enabled="true"> <stringProp name="TestPlan.comments">测试</stringProp> <stringProp name="ThreadGroup.on_sample_error">continue</stringProp> <elementProp name="ThreadGroup.main_controller" elementType="LoopController" guiclass="LoopControlPanel" testclass="LoopController" testname="循环控制器" enabled="true"> <boolProp name="LoopController.continue_forever">false</boolProp> <stringProp name="LoopController.loops">50</stringProp> </elementProp> <stringProp name="ThreadGroup.num_threads">2</stringProp> <stringProp name="ThreadGroup.ramp_time">1</stringProp> <longProp name="ThreadGroup.start_time">1496464278000</longProp> <longProp name="ThreadGroup.end_time">1496464278000</longProp> <boolProp name="ThreadGroup.scheduler">false</boolProp> <stringProp name="ThreadGroup.duration"></stringProp> <stringProp name="ThreadGroup.delay"></stringProp> </ThreadGroup> <hashTree> <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="HTTP请求" enabled="true"> <elementProp name="HTTPsampler.Arguments" elementType="Arguments" guiclass="HTTPArgumentsPanel" testclass="Arguments" testname="用户定义的变量" enabled="true"> <collectionProp name="Arguments.arguments"/> </elementProp> <stringProp name="HTTPSampler.domain">lansgg.blog.51cto.com</stringProp> <stringProp name="HTTPSampler.port">80</stringProp> <stringProp name="HTTPSampler.connect_timeout"></stringProp> <stringProp name="HTTPSampler.response_timeout"></stringProp> <stringProp name="HTTPSampler.protocol"></stringProp> <stringProp name="HTTPSampler.contentEncoding"></stringProp> <stringProp name="HTTPSampler.path"></stringProp> <stringProp name="HTTPSampler.method">GET</stringProp> <boolProp name="HTTPSampler.follow_redirects">true</boolProp> <boolProp name="HTTPSampler.auto_redirects">false</boolProp> <boolProp name="HTTPSampler.use_keepalive">true</boolProp> <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp> <stringProp name="HTTPSampler.implementation">HttpClient4</stringProp> <boolProp name="HTTPSampler.monitor">false</boolProp> <stringProp name="HTTPSampler.embedded_url_re"></stringProp> <stringProp name="TestPlan.comments">HTTP请求</stringProp> </HTTPSamplerProxy> <hashTree/> <ResultCollector guiclass="ViewResultsFullVisualizer" testclass="ResultCollector" testname="察看结果树" enabled="true"> <boolProp name="ResultCollector.error_logging">false</boolProp> <objProp> <name>saveConfig</name> <value class="SampleSaveConfiguration"> <time>true</time> <latency>true</latency> <timestamp>true</timestamp> <success>true</success> <label>true</label> <code>true</code> <message>true</message> <threadName>true</threadName> <dataType>true</dataType> <encoding>false</encoding> <assertions>true</assertions> <subresults>true</subresults> <responseData>false</responseData> <samplerData>false</samplerData> <xml>false</xml> <fieldNames>false</fieldNames> <responseHeaders>false</responseHeaders> <requestHeaders>false</requestHeaders> <responseDataOnError>false</responseDataOnError> <saveAssertionResultsFailureMessage>false</saveAssertionResultsFailureMessage> <assertionsResultsToSave>0</assertionsResultsToSave> <bytes>true</bytes> </value> </objProp> <stringProp name="TestPlan.comments">察看结果树</stringProp> <stringProp name="filename"></stringProp> </ResultCollector> <hashTree/> </hashTree> </hashTree> </hashTree> </jmeterTestPlan>
当我们在linux终端上执行时,结果如下,
[root@vm6 rpc]# /usr/local/apache-jmeter-3.2/bin/jmeter -n -t test.jmx -l text.jtl Creating summariser <summary> Created the tree successfully using test.jmx Starting the test @ Sat Jun 03 20:41:25 CST 2017 (1496493685813) Waiting for possible Shutdown/StopTestNow/Heapdump message on port 4445 summary + 46 in 00:00:18 = 2.5/s Avg: 2745 Min: 423 Max: 3135 Err: 0 (0.00%) Active: 49 Started: 50 Finished: 1 summary + 54 in 00:00:08 = 7.2/s Avg: 2018 Min: 413 Max: 8000 Err: 0 (0.00%) Active: 0 Started: 50 Finished: 50 summary = 100 in 00:00:26 = 3.9/s Avg: 2353 Min: 413 Max: 8000 Err: 0 (0.00%) Tidying up ... @ Sat Jun 03 20:42:07 CST 2017 (1496493727553) ... end of run
而我们要将这些结果实时的在rpc client展示出来
接口文件:
syntax = "proto3"; package rpc; service RPC { // send config jmx rpc sendConfFile(Content) returns (Status) {} // run jmeter test rpc runJMeter(Content) returns (stream Content) {} } message Content { string text = 1; } message Status { int64 code = 1; }
server端:
# -*- encoding=utf-8 -*- from concurrent import futures import time import subprocess import codecs import sys import os import json import logging import grpc import rpc_pb2 import rpc_pb2_grpc logging.basicConfig(level=logging.DEBUG, format=‘%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s‘, datefmt=‘%a, %d %b %Y %H:%M:%S‘, filename=‘rpc_server.log‘, filemode=‘a+‘) _ONE_DAY_IN_SECONDS = 60 * 60 * 24 jmeter_config = os.path.join(os.getcwd(),r‘conf/config.jmx‘) jmeter_path = r‘/usr/local/apache-jmeter-3.2/bin/jmeter‘ jmeter_result = os.path.join(os.getcwd(),r‘result/result.jtl‘) class Performance(rpc_pb2_grpc.RPCServicer): def sendConfFile(self, content,context): ‘‘‘ 保存配置文件,如config.jmx ‘‘‘ text = content.text try: conf_handle = codecs.open(jmeter_config,‘w‘,encoding=‘utf-8‘) conf_handle.write(text) logging.info("sendConfFile Success!") return rpc_pb2.Status(code=0) except Exception,e: print e return rpc_pb2.Status(code=1) def runJMeter(self, content,context): logging.info("begin runJmeter.") ‘‘‘执行测试任务,并将实时结果返回‘‘‘ iplist = content.text if iplist: cmd = jmeter_path + " -n -t " + jmeter_config + " -l " + jmeter_result + " -R " + iplist else: cmd = jmeter_path + " -n -t " + jmeter_config + " -l " + jmeter_result logging.info(cmd) popen = subprocess.Popen(cmd, stdout=subprocess.PIPE, universal_newlines=True,shell=True) for stdout_line in iter(popen.stdout.readline, ""): log_line = rpc_pb2.Content(text=stdout_line) yield log_line popen.stdout.close() return_code = popen.wait() if return_code: logging.warn(return_code) raise subprocess.CalledProcessError(return_code, cmd) def serve(): server = grpc.server(futures.ThreadPoolExecutor(max_workers=10)) rpc_pb2_grpc.add_RPCServicer_to_server(Performance(),server) server.add_insecure_port(‘[::]:50051‘) server.start() try: while True: time.sleep(_ONE_DAY_IN_SECONDS) except KeyboardInterrupt: server.stop(0) if __name__ == ‘__main__‘: serve()
client端:
# -*- encoding=utf-8 -*- from concurrent import futures import time import subprocess import codecs import sys import os import logging import json import grpc import rpc_pb2 import rpc_pb2_grpc logging.basicConfig(level=logging.DEBUG, format=‘%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s‘, datefmt=‘%a, %d %b %Y %H:%M:%S‘, filename=‘rpc.log‘, filemode=‘a+‘) rpc_server = r‘192.168.18.128‘ rpc_port = ‘50051‘ jmeter_config = os.path.join(os.getcwd(),r‘conf/config.jmx‘) jmeter_path = r‘/usr/local/apache-jmeter-3.2/bin/jmeter‘ jmeter_result = os.path.join(os.getcwd(),r‘conf/result.jtl‘) class performance(): ‘‘‘性能测试的客户端接口‘‘‘ def __init__(self,ip,port): ‘‘‘初始化,连接RPC服务‘‘‘ logging.info("performance_client init") conn = grpc.insecure_channel(ip + ‘:‘ + port) self.stub_client = rpc_pb2_grpc.RPCStub(channel=conn) def sendConfig(self,filename): file_handle = codecs.open(filename, ‘r‘, encoding=‘utf-8‘) content = file_handle.read() ‘‘‘向RPC server发送测试的配置文件‘‘‘ response = self.stub_client.sendConfFile(rpc_pb2.Content(text=content)) return response.code def runJmeter(self,iplist): ‘‘‘运行测试 返回一个生成器,内容为测试过程中的实时输出‘‘‘ content = iplist logging.info("iplist %s" %content) for log in self.stub_client.runJMeter(rpc_pb2.Content(text=content)): yield log if __name__ == ‘__main__‘: client = performance(rpc_server,rpc_port) code = client.sendConfig(‘test.jmx‘) iplist = r‘10.1.1.1,10.1.1.2‘ for real_time_results in client.runJmeter(None): print "get realtime log from server : %s" % real_time_results.text
结果:
5、基于grcp协议获取jmeter最终压测报告,并将报告保存至client端
6、压测中途停止jmeter
接口文件
syntax = "proto3"; package rpc; service RPC { // send config jmx rpc sendConfFile(Content) returns (Status) {} // run jmeter test rpc runJMeter(Content) returns (stream Content) {} // generateResult rpc generateResult(empty) returns (stream Content) {} // getResult rpc getResult(empty) returns (stream Content) {} // getResult rpc stopJMeter(empty) returns (stream Content) {} } message Content { string text = 1; } message Status { int64 code = 1; } message empty { }
server端:
# -*- encoding=utf-8 -*- from concurrent import futures import time import subprocess import codecs import sys import os import json import logging import grpc import rpc_pb2 import rpc_pb2_grpc logging.basicConfig(level=logging.DEBUG, format=‘%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s‘, datefmt=‘%a, %d %b %Y %H:%M:%S‘, filename=‘rpc_server.log‘, filemode=‘a+‘) _ONE_DAY_IN_SECONDS = 60 * 60 * 24 jmeter_config = os.path.join(os.getcwd(),r‘conf/config.jmx‘) jmeter_path = r‘/usr/local/apache-jmeter-3.2/bin/jmeter‘ jmeter_result_file = os.path.join(os.getcwd(),r‘result/result.jtl‘) jmeter_result_dir = os.path.join(os.getcwd(),r‘result/summary/‘) class Performance(rpc_pb2_grpc.RPCServicer): def sendConfFile(self, content,context): ‘‘‘ 保存配置文件,如config.jmx ‘‘‘ text = content.text try: conf_handle = codecs.open(jmeter_config,‘w‘,encoding=‘utf-8‘) conf_handle.write(text) logging.info("sendConfFile Success!") return rpc_pb2.Status(code=0) except Exception,e: print e return rpc_pb2.Status(code=1) def runJMeter(self, content,context): logging.info("begin runJmeter.") ‘‘‘执行测试任务,并将实时结果返回‘‘‘ iplist = content.text if iplist: cmd = jmeter_path + " -n -t " + jmeter_config + " -l " + jmeter_result_file + " -R " + iplist else: cmd = jmeter_path + " -n -t " + jmeter_config + " -l " + jmeter_result_file logging.info(cmd) popen = subprocess.Popen(cmd, stdout=subprocess.PIPE, universal_newlines=True,shell=True) for stdout_line in iter(popen.stdout.readline, ""): log_line = rpc_pb2.Content(text=stdout_line) yield log_line popen.stdout.close() return_code = popen.wait() if return_code: logging.warn(return_code) raise subprocess.CalledProcessError(return_code, cmd) def generateResult(self,empty,context): ‘‘‘调用jmeter,生成汇总测试结果 ‘‘‘ cmd = jmeter_path + " -g " + jmeter_result_file + " -o " + jmeter_result_dir logging.info(cmd) popen = subprocess.Popen(cmd, stdout=subprocess.PIPE,shell=True) return_code = popen.wait() return rpc_pb2.Status(code=return_code) def getResult(self, empty, content): ‘‘‘获取汇总测试结果,返回给客户端 ‘‘‘ summary_file = os.path.join(jmeter_result_dir,r"content/js/dashboard.js") file_handle = codecs.open(summary_file,‘r‘,encoding=‘utf-8‘) for line in file_handle.readlines(): yield rpc_pb2.Content(text=line) file_handle.close() def stopJMeter(self, empty, content): ‘‘‘杀死正在执行的任务‘‘‘ os.system("ps -ef | grep jmeter | grep -v grep | awk ‘{print $2}‘| xargs kill -9") return rpc_pb2.Status(code=1) def serve(): server = grpc.server(futures.ThreadPoolExecutor(max_workers=10)) rpc_pb2_grpc.add_RPCServicer_to_server(Performance(),server) server.add_insecure_port(‘[::]:50051‘) server.start() try: while True: time.sleep(_ONE_DAY_IN_SECONDS) except KeyboardInterrupt: server.stop(0) if __name__ == ‘__main__‘: serve()
client端:
# -*- encoding=utf-8 -*- from concurrent import futures import time import subprocess import codecs import sys import os import logging import json import grpc import rpc_pb2 import rpc_pb2_grpc logging.basicConfig(level=logging.DEBUG, format=‘%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s‘, datefmt=‘%a, %d %b %Y %H:%M:%S‘, filename=‘rpc.log‘, filemode=‘a+‘) rpc_server = r‘192.168.18.128‘ rpc_port = ‘50051‘ jmeter_config = os.path.join(os.getcwd(),r‘conf/config.jmx‘) jmeter_path = r‘/usr/local/apache-jmeter-3.2/bin/jmeter‘ jmeter_result = os.path.join(os.getcwd(),r‘conf/result.jtl‘) class performance(): ‘‘‘性能测试的客户端接口‘‘‘ def __init__(self,ip,port): ‘‘‘初始化,连接RPC服务‘‘‘ logging.info("performance_client init") conn = grpc.insecure_channel(ip + ‘:‘ + port) self.stub_client = rpc_pb2_grpc.RPCStub(channel=conn) def sendConfig(self,filename): file_handle = codecs.open(filename, ‘r‘, encoding=‘utf-8‘) content = file_handle.read() ‘‘‘向RPC server发送测试的配置文件‘‘‘ response = self.stub_client.sendConfFile(rpc_pb2.Content(text=content)) return response.code def runJmeter(self,iplist): ‘‘‘运行测试 返回一个生成器,内容为测试过程中的实时输出‘‘‘ content = iplist logging.info("iplist %s" %content) for log in self.stub_client.runJMeter(rpc_pb2.Content(text=content)): yield log def generateResult(self): ‘‘‘在rpc server 端生成 jmeter 最终报告‘‘‘ response = self.stub_client.generateResult(rpc_pb2.empty()) return response.code def getResult(self, local_file): ‘‘‘从RPC server回传测试结果,保存到本地文件local_file‘‘‘ file_handle = codecs.open(local_file, ‘w‘, encoding=‘utf-8‘) for line in self.stub_client.getResult(rpc_pb2.empty()): file_handle.write(line.text) file_handle.close() def stopJmeter(self): ‘‘‘终止测试任务 ‘‘‘ self.stub_client.stopJMeter(rpc_pb2.empty()) if __name__ == ‘__main__‘: client = performance(rpc_server,rpc_port) code = client.sendConfig(‘test.jmx‘) iplist = r‘10.1.1.1,10.1.1.2‘ for real_time_results in client.runJmeter(None): print "get realtime log from server : %s" % real_time_results.text print client.generateResult() print client.getResult(‘2222222222222222‘) client.stopJmeter()
结果:
当你想停止jmeter压测,调用stop即可。
本文出自 “大風” 博客,请务必保留此出处http://lansgg.blog.51cto.com/5675165/1931961
python grpc 应用