python grpc 应用
短信预约 -IT技能 免费直播动态提醒
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.protoserver 端代码：
# -*- 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即可。
免责声明：
① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的，并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据，供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。
② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341
