简介:平时测试的时候,并不是所有的接口都是页面端的,不是页面端的也就不是http请求了,很多时候使用的是duboo请求,做接口测试的时候,或者做自动化和性能测试的时候,我们是需要使用dubbo接口了,在企业中我做了很好的封装,但是因为保密的原因企业上的东西不能上传的网上,所以只能上次个简陋版的,虽说是简陋版的,但是也足够应付大部分的场合了。
拓展:接口测试的时候为了能闭环,需要使用pymysql进行查询数据库,校验数据是否正确落库。一般数据正常落库即可闭环,当然如果是特殊场景则根据具体的场景进行闭环操作
这个是参考的模型:
”’
方法调用案例:
conn = InvokeDubboApi(‘127.0.0.1:8080’)
data = {
‘dubbo_service’: ‘xxx.xxx.xxxx.xxxx.xxxx.xxxx.Service’,
‘dubbo_method’: ‘xxxxx’,
‘parameters’: ({“age”:41,”name”:”tom”},”sh”,564645,)
}
invoke = json.loads(conn.invoke_dubbo_api(data))
conn.logout()
”’
以下为具体的代码:
import re
import telnetlib
import time
import logging
#获取日志等级
logging.basicConfig(level = logging.INFO,format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
"""通过telnet连接dubbo服务,调用dubbo接口"""
class TelnetClient(object):
def __init__(self, server_host, server_port):
self.conn = telnetlib.Telnet()
self.server_host = server_host
self.server_port = server_port
# telnet登录主机
def connect_dubbo(self):
try:
logging.info("telent连接dubbo服务端: 地址和端口分别为 {} {}".format(self.server_host, self.server_port))
self.conn.open(self.server_host, port=self.server_port)
return True
except Exception as e:
logging.info('连接失败, 原因是: {}'.format(str(e)))
return False
# 执行传过来的命令,并输出其执行结果
def execute_command(self, command):
# 执行命令
cmd = 'invoke {}\n'.format(command).encode("utf-8")
self.conn.write(cmd)
# 初始化调用次数
invoke_count = 0
# 若调用无返回时,以下重发机制记录次数并重试
result = self.conn.read_very_eager().decode(encoding='utf-8').split('\r\n')[0]
while result == '':
time.sleep(1)
result = self.conn.read_very_eager().decode(encoding='utf-8').split('\r\n')[0]
invoke_count += 1
if invoke_count>=5:
logging.info("调用dubbo接口超过五次,调用失败")
return '调用dubbo接口失败'
return result
# 退出telnet
def logout_host(self):
self.conn.write(b"exit\n")
logging.info("登出成功")
'''此类用于执行dubbo接口'''
class InvokeDubboApi(object):
def __init__(self, content):
#解析dubbo部署的ip和port
try:
dubboaddrre = re.compile(r"([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+:[0-9]+)", re.I)
result = dubboaddrre.search(str(content)).group()
server_host = result.split(":")[0]
server_port = result.split(":")[1]
logging.info("获取到dubbo部署信息" + result)
except Exception as e:
raise Exception("获取dubbo部署信息失败:{}".format(e))
try:
self.telnet_client = TelnetClient(server_host, server_port)
self.login_flag = self.telnet_client.connect_dubbo()
except Exception as e:
logging.info("invokedubboapi init error" + e)
#调用dubbo接口
def invoke_dubbo_api(self, params):
cmd = params[0] + '.' +params[1]+ '({'+"name:{},".format(params[2])+"age:{}".format(params[3])+'})'
logging.info("调用命令是:{}".format(cmd))
try:
if self.login_flag:
result= self.telnet_client.execute_command(cmd)
assert "操作成功" in result,"开发怎么回事,这点小事都做不好么?"
logging.info("接口响应是,result={}".format(result))
return result
else:
logging.info("登陆失败")
except Exception as e:
raise Exception("调用接口异常, 接口响应是result={}, 异常信息为:{}".format(result, e))
self.logout()
# 调用多个dubbo接口,注:确保所有接口是同一个ip和port
def invoke_dubbo_apis(self,datas):
summary = []
if isinstance(datas,list):
for i in range(len(datas)):
result = self.invoke_dubbo_api(datas[i])
summary.append({"data":datas[i],"result":result})
return summary
else:
return "请确认入参是list"
def logout(self):
self.telnet_client.logout_host()
if __name__ == '__main__':
#前两个参数为接口和方法,另外两个参数为方法入参
params = {'xxx.xxx.xx.xxxx.xxxx.xxxxService','xxxxxadd',"江云",27}
i = InvokeDubboApi('127.0.0.1:8080') #dubbo接口对应的服务器IP和端口
i.invoke_dubbo_api(params)
i.logout()
Dubbo的介绍:
Dubbo 的诞生和 SOA 分布式架构的流行有着莫大的关系。SOA 面向服务的架构(Service Oriented Architecture),也就是把工程按照业务逻辑拆分成服务层、表现层两个工程。服务层中包含业务逻辑,只需要对外提供服务即可。表现层只需要处理和页面的交互,业务逻辑都是调用服务层的服务来实现。SOA架构中有两个主要角色:服务提供者(Provider)和服务使用者(Consumer)。
企业中使用Dubbo的好处有:
负载均衡:
同一个服务部署在不同的机器时该调用那一台机器上的服务
容错机制:
随着系统的发展,服务越来越多,服务间依赖关系变得错踪复杂,甚至分不清哪个应用要在哪个应用之前启动。Dubbo 可以为我们解决服务之间互相是如何调用的。
服务治理:
基于访问压力实时管理集群容量,提高集群利用率。
服务降级:
某个服务挂掉之后调用备用服务
另外,Dubbo 除了能够应用在分布式系统中,也可以应用在现在比较火的微服务系统中。
版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。
如需转载请保留出处:https://bianchenghao.cn/36685.html